postgresqlsqlalchemypostgresql-14

SQLAlchemy create unique constraint with the hash value of multiple columns


I want to create a UniqueConstraint using a hash of multiple columns.
Postgresql 15 provides the hash function hash_record_extended, but I couldn't figure out the syntax.

__table_args__ = (
    UniqueConstraint(
        func.hash_record_extended(
            (
                "col1",
                "col2",
                "col3",
            ),
            0, 
        ),
    ),
)

The above gives the error:

sqlalchemy.exc.ArgumentError: Can't add unnamed column to column collection


Solution

  • Edit: As per this sqlalchemy discussion functions in UniqueConstraint is not supported. I am leaving a solution that uses unique index instead if that helps. You could check Postgres unique constraint vs index for some explanation.

    I am not really sure if hash_record_extended is a public facing api as I can not seem to find in the docs (will edit this appropriately if anyone can find a source). That being said, you can create a normal index and make it unique like so. This will raise IntegrityError if you attempt to insert non unique values.

    Index(
        "some_name",
        func.hash_record_extended(func.row(col1, col2, col3), 0),
        unique=True,
    )
    

    This generates the following sql

    CREATE UNIQUE INDEX some_name ON some_table (
      hash_record_extended(
        row(col1, col2, col3), 
        0
      )
    )
    
    

    Complete copy-pasteable code using SQLAlchemy 2

    from sqlalchemy import create_engine, func, Index
    from sqlalchemy.orm import Mapped, DeclarativeBase, mapped_column
    
    class Base(DeclarativeBase):
        pass
    
    class SomeTable(Base):
        __tablename__ = "some_table"
        id: Mapped[int] = mapped_column(primary_key=True)
        col1: Mapped[str] = mapped_column()
        col2: Mapped[str] = mapped_column()
        col3: Mapped[str] = mapped_column()
        __table_args__ = (
            Index(
                "some_name",
                func.hash_record_extended(func.row(col1, col2, col3), 0),
                unique=True,
            ),
        )
    
    connection_string = "postgresql+psycopg://"
    engine = create_engine(connection_string, echo=True)
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)