Re: [sqlalchemy] Unique constraint error in PostgrSQL when migrating with Alembic

2023-09-13 Thread Mike Bayer
your revisions table has a composite unique constraint - one constraint with 
two columns in it.   therefore to refer to this constraint via foreign key, you 
need a single composite foreignkeyconstraint - again, one constraint that links 
two columns together.  you would use ForeignKeyConstraint for this, not 
ForeignKey, see ForeignKeyConstraint in 
https://docs.sqlalchemy.org/en/20/core/constraints.html#defining-foreign-keys .

On Wed, Sep 13, 2023, at 1:43 PM, Lord Wolfenstein wrote:
> I have a database that looks like this that I create with Alembic. The 
> relationships are trivial except between Revision and ObjectCount, there it 
> used two foreign keys
> aaa.png
> 
> The code looks like this 
> 
> 
> 
> from datetime import datetime
> from typing import Optional
> 
> from sqlalchemy import ForeignKey, create_engine, UniqueConstraint
> from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, 
> relationship, sessionmaker  # type: ignore
> 
> 
> class Base(DeclarativeBase):
> pass
> 
> 
> class Blueprint(Base):
> __tablename__ = "blueprints"
> 
> blueprint_id: Mapped[int] = mapped_column(primary_key=True)
> filename: Mapped[str]
> created: Mapped[datetime] = mapped_column(default=datetime.utcnow)
> 
> revisions: Mapped[list["Revision"]] = 
> relationship(back_populates="blueprint")  # type: ignore
> 
> def get_new_revision_number(self) -> int:
> if not self.revisions:
> return 1
> return max(revision.revision_number for revision in self.revisions) + 
> 1
> 
> def __str__(self):
> return f"{self.filename} : {self.blueprint_id}"
> 
> 
> class Revision(Base):
> __tablename__ = "revisions"
> __table_args__ = (
> UniqueConstraint("blueprint_id", "revision_number", 
> name="revision_blueprint_revision_number"),
> )
> 
> blueprint_id: Mapped[int] = 
> mapped_column(ForeignKey("blueprints.blueprint_id"), primary_key=True)
> revision_number: Mapped[int] = mapped_column(primary_key=True)
> date: Mapped[datetime] = mapped_column(default=datetime.utcnow)
> savedata: Mapped[str]
> 
> blueprint: Mapped["Blueprint"] = relationship(back_populates="revisions") 
>  # type: ignore
> object_count: Mapped[Optional[list["ObjectCount"]]] = relationship(
> back_populates="revision",
> primaryjoin="and_(Revision.blueprint_id==ObjectCount.blueprint_id, 
> Revision.revision_number==ObjectCount.revision_number)",
> )  # type: ignore
> 
> def __str__(self):
> return f"{self.blueprint.filename} : {self.blueprint_id} : 
> {self.revision_number}"
> 
> 
> class Object(Base):
> __tablename__ = "objects"
> 
> object_id: Mapped[int] = mapped_column(primary_key=True)
> name: Mapped[str]
> 
> def __str__(self):
> return self.name
> 
> 
> # https://docs.sqlalchemy.org/en/20/orm/join_conditions.html
> # https://docs.sqlalchemy.org/en/20/orm/basic_relationships.html
> # 
> https://docs.sqlalchemy.org/en/20/core/constraints.html#sqlalchemy.schema.UniqueConstraint
> class ObjectCount(Base):
> __tablename__ = "object_count"
> __table_args__ = (
> UniqueConstraint("blueprint_id", "revision_number", "object_id", 
> name="o_c_unique"),
> )
> 
> blueprint_id: Mapped[int] = 
> mapped_column(ForeignKey("revisions.blueprint_id"), primary_key=True)
> revision_number: Mapped[int] = 
> mapped_column(ForeignKey("revisions.revision_number"), primary_key=True)
> object_id: Mapped[int] = mapped_column(ForeignKey("objects.object_id"), 
> primary_key=True)
> 
> count: Mapped[int]
> 
> object: Mapped["Object"] = relationship()  # type: ignore
> revision: Mapped["Revision"] = relationship(
> back_populates="object_count",
> primaryjoin="and_(Revision.blueprint_id==ObjectCount.blueprint_id, 
> Revision.revision_number==ObjectCount.revision_number)",
> )  # type: ignore
> 
> def __str__(self):
> return f"{self.revision.blueprint.filename} {self.revision_number} 
> {self.object.name} {self.count}"
> 
> 
> DATABASE = "postgresql+psycopg://user:password@192.168.10.111:5432/mydatabase"
> DATABASE = 
> "postgresql+psycopg2://user:password@192.168.10.111:5432/mydatabase"
> #DATABASE = "sqlite:///database.db"
> engine = create_engine(DATABASE)
> SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
> 
> 
> I can create migrations with Alembic no problem. And I can migrate when I use 
> SQLite. But when I try to migrate with PostgreSQL I get this error:
> 
> sqlalchemy.exc.ProgrammingError: (psycopg2.errors.InvalidForeignKey) there is 
> no unique constraint matching given keys for referenced table "revisions"
> 
> [SQL:
> CREATE TABLE object_count (
> blueprint_id INTEGER NOT NULL,
> revision_number INTEGER NOT NULL,
> object_id INTEGER NOT NULL,
> count INTEGER NOT NULL,
> PRIMARY KEY (blueprint_id, 

[sqlalchemy] Unique constraint error in PostgrSQL when migrating with Alembic

2023-09-13 Thread Lord Wolfenstein
I have a database that looks like this that I create with Alembic. The 
relationships are trivial except between Revision and ObjectCount, there it 
used two foreign keys
[image: aaa.png]

The code looks like this 



from datetime import datetime
from typing import Optional

from sqlalchemy import ForeignKey, create_engine, UniqueConstraint
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, 
relationship, sessionmaker  # type: ignore


class Base(DeclarativeBase):
pass


class Blueprint(Base):
__tablename__ = "blueprints"

blueprint_id: Mapped[int] = mapped_column(primary_key=True)
filename: Mapped[str]
created: Mapped[datetime] = mapped_column(default=datetime.utcnow)

revisions: Mapped[list["Revision"]] = 
relationship(back_populates="blueprint")  # type: ignore

def get_new_revision_number(self) -> int:
if not self.revisions:
return 1
return max(revision.revision_number for revision in self.revisions) 
+ 1

def __str__(self):
return f"{self.filename} : {self.blueprint_id}"


class Revision(Base):
__tablename__ = "revisions"
__table_args__ = (
UniqueConstraint("blueprint_id", "revision_number", 
name="revision_blueprint_revision_number"),
)

blueprint_id: Mapped[int] = 
mapped_column(ForeignKey("blueprints.blueprint_id"), primary_key=True)
revision_number: Mapped[int] = mapped_column(primary_key=True)
date: Mapped[datetime] = mapped_column(default=datetime.utcnow)
savedata: Mapped[str]

blueprint: Mapped["Blueprint"] = 
relationship(back_populates="revisions")  # type: ignore
object_count: Mapped[Optional[list["ObjectCount"]]] = relationship(
back_populates="revision",
primaryjoin="and_(Revision.blueprint_id==ObjectCount.blueprint_id, 
Revision.revision_number==ObjectCount.revision_number)",
)  # type: ignore

def __str__(self):
return f"{self.blueprint.filename} : {self.blueprint_id} : 
{self.revision_number}"


class Object(Base):
__tablename__ = "objects"

object_id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]

def __str__(self):
return self.name


# https://docs.sqlalchemy.org/en/20/orm/join_conditions.html
# https://docs.sqlalchemy.org/en/20/orm/basic_relationships.html
# 
https://docs.sqlalchemy.org/en/20/core/constraints.html#sqlalchemy.schema.UniqueConstraint
class ObjectCount(Base):
__tablename__ = "object_count"
__table_args__ = (
UniqueConstraint("blueprint_id", "revision_number", "object_id", 
name="o_c_unique"),
)

blueprint_id: Mapped[int] = 
mapped_column(ForeignKey("revisions.blueprint_id"), primary_key=True)
revision_number: Mapped[int] = 
mapped_column(ForeignKey("revisions.revision_number"), primary_key=True)
object_id: Mapped[int] = mapped_column(ForeignKey("objects.object_id"), 
primary_key=True)

count: Mapped[int]

object: Mapped["Object"] = relationship()  # type: ignore
revision: Mapped["Revision"] = relationship(
back_populates="object_count",
primaryjoin="and_(Revision.blueprint_id==ObjectCount.blueprint_id, 
Revision.revision_number==ObjectCount.revision_number)",
)  # type: ignore

def __str__(self):
return f"{self.revision.blueprint.filename} {self.revision_number} 
{self.object.name} {self.count}"


DATABASE = 
"postgresql+psycopg://user:password@192.168.10.111:5432/mydatabase"
DATABASE = 
"postgresql+psycopg2://user:password@192.168.10.111:5432/mydatabase"
#DATABASE = "sqlite:///database.db"
engine = create_engine(DATABASE)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


I can create migrations with Alembic no problem. And I can migrate when I 
use SQLite. But when I try to migrate with PostgreSQL I get this error:

sqlalchemy.exc.ProgrammingError: (psycopg2.errors.InvalidForeignKey) there 
is no unique constraint matching given keys for referenced table "revisions"

[SQL:
CREATE TABLE object_count (
blueprint_id INTEGER NOT NULL,
revision_number INTEGER NOT NULL,
object_id INTEGER NOT NULL,
count INTEGER NOT NULL,
PRIMARY KEY (blueprint_id, revision_number, object_id),
FOREIGN KEY(blueprint_id) REFERENCES revisions (blueprint_id),
FOREIGN KEY(object_id) REFERENCES objects (object_id),
FOREIGN KEY(revision_number) REFERENCES revisions (revision_number),
CONSTRAINT o_c_unique UNIQUE (blueprint_id, revision_number, 
object_id)
)
]

I think I clearly see a unique constraint in the code. I add the 
UniqueConstraint in __table_args__ but PostgreSQL still hits the wall then 
I migrate. I get the same error with both psycopg2==2.9.6 and 
psycopg==3.1.10. What frustrates me is that it works and all tests pass 
when I use SQLite.

-- 
SQLAlchemy - 
The Python SQL Toolkit and Object Relational Mapper

http://www.sqlalchemy.org/

To post example code,