pythonpostgresqlsqlalchemyflask-sqlalchemy

Case-insensitive UniqueConstraint using SQLAlchemy with PostgreSQL DB


I am using SQLAlchemy and PostgreSQL, and I am trying to create a case-insensitive unique constraint that works like this

        UniqueConstraint(
            'country',
            'organisation_id',
            func.lower(func.trim('name')),
            name='uq_country_orgid_lower_trim_name'
        )

Ensuring a unique combination of name, country and organisation id, regardless of case and spacing in the name, i.e. "Name 1", "name1", "nAmE 1" would all be handled as "name1" in the check.

I want to make sure that I do not change the actual case or spacing of the name saved in the database.

How would I go about this?


Solution

  • This is mostly the ORM version of @snakecharmerb and avoids the use of text (nothing wrong with sanitized text, it is just a preference). This answer suggests using a unique index, instead of unique constraint because SQLA lacks support.

    from sqlalchemy import create_engine, func, Index, select, column
    from sqlalchemy.exc import IntegrityError
    from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
    
    class Base(DeclarativeBase): ...
    
    class SomeTable(Base):
        __tablename__ = "some_table"
        __table_args__ = (
            Index(
                "uq_country_orgid_lower_trim_name",
                "country",
                "organization_id",
                func.lower(func.replace(column("name"), " ", "")),
                unique=True,
            ),
        )
        name: Mapped[str]
        id: Mapped[int] = mapped_column(primary_key=True)
        country: Mapped[str]
        organization_id: Mapped[int]
    
    engine = create_engine("postgresql+psycopg://")
    Base.metadata.create_all(engine)
    
    with Session(engine) as session:
        session.add(SomeTable(country="C1", organization_id=1, name="NAME 1"))
        session.commit()
    
    with Session(engine) as session:
        rows = (
            SomeTable(country="C1", organization_id=1, name="NAME1"),
            SomeTable(country="C1", organization_id=1, name="Name 1"),
            SomeTable(country="C1", organization_id=1, name="name1"),
            SomeTable(country="C1", organization_id=1, name="nAmE 1"),
        )
        for row in rows:
            try:
                with session.begin_nested():
                    session.add(row)
            except IntegrityError:
                print("failed as expected")
    
    with Session(engine) as session:
        row = session.scalar(select(SomeTable))
        assert row
        assert row.name == "NAME 1"