Sqlalchemy Unique Across Multiple Columns

sqlalchemy unique across multiple columns

Extract from the documentation of the Column:

unique – When True, indicates that this column contains a unique
constraint, or if index is True as well, indicates that the Index
should be created with the unique flag. To specify multiple columns in
the constraint/index or to specify an explicit name, use the
UniqueConstraint or Index constructs explicitly.

As these belong to a Table and not to a mapped Class, one declares those in the table definition, or if using declarative as in the __table_args__:

# version1: table definition
mytable = Table('mytable', meta,
    # ...
    Column('customer_id', Integer, ForeignKey('customers.customer_id')),
    Column('location_code', Unicode(10)),

    UniqueConstraint('customer_id', 'location_code', name='uix_1')
    )
# or the index, which will ensure uniqueness as well
Index('myindex', mytable.c.customer_id, mytable.c.location_code, unique=True)

# version2: declarative
class Location(Base):
    __tablename__ = 'locations'
    id = Column(Integer, primary_key = True)
    customer_id = Column(Integer, ForeignKey('customers.customer_id'), nullable=False)
    location_code = Column(Unicode(10), nullable=False)
    __table_args__ = (UniqueConstraint('customer_id', 'location_code', name='_customer_location_uc'),
                     )

Unique constraint on multiple columns in sqlalchemy

Probably the easiest solution is to add a check constraint for c1 < c2 (or c1 <= c2 if they're allowed to be the same) so that (c1, c2) will always be in "ascending order":

import sqlalchemy as sa

connection_uri = (
    "mssql+pyodbc://@localhost:49242/myDb?driver=ODBC+Driver+17+for+SQL+Server"
)
engine = sa.create_engine(connection_uri)
Base = declarative_base()

class So64232358(Base):
    __tablename__ = "so64232358"
    id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    c1 = sa.Column(sa.Integer, nullable=False)
    c2 = sa.Column(sa.Integer, nullable=False)
    comment = sa.Column(sa.String(50))
    sa.CheckConstraint(c1 < c2)
    sa.UniqueConstraint(c1, c2)

Base.metadata.drop_all(engine, checkfirst=True)
Base.metadata.create_all(engine)
"""SQL rendered:
CREATE TABLE so64232358 (
    id INTEGER NOT NULL IDENTITY, 
    c1 INTEGER NOT NULL, 
    c2 INTEGER NOT NULL, 
    comment VARCHAR(50) NULL, 
    PRIMARY KEY (id), 
    CHECK (c1 < c2), 
    UNIQUE (c1, c2)
)
"""

Session = sessionmaker(bind=engine)
session = Session()

obj = So64232358(c1=2, c2=1, comment="no es bueno")
session.add(obj)
try:
    session.commit()
except sa.exc.IntegrityError as ie:
    print(ie)
    """console output:
    (pyodbc.IntegrityError) ('23000', '[23000] [Microsoft]
    [ODBC Driver 17 for SQL Server][SQL Server]The INSERT statement conflicted
    with the CHECK constraint "CK__so64232358__429B0397". The conflict
    occurred in database "myDb", table "dbo.so64232358".
    (547) (SQLExecDirectW);
    [23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]
    The statement has been terminated. (3621)')
    """
    session.rollback()
obj = So64232358(c1=1, c2=2, comment="bueno")
session.add(obj)
session.commit()  # no error
obj = So64232358(c1=1, c2=2, comment="duplicado")
session.add(obj)
try:
    session.commit()
except sa.exc.IntegrityError as ie:
    print(ie)
    """console output:
    (pyodbc.IntegrityError) ('23000', "[23000] [Microsoft]
    [ODBC Driver 17 for SQL Server][SQL Server]Violation of UNIQUE KEY
    constraint 'UQ__so642323__E13250592117193A'. Cannot insert duplicate key
    in object 'dbo.so64232358'. The duplicate key value is (1, 2).
    (2627)(SQLExecDirectW);
    [23000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]
    The statement has been terminated. (3621)")
    """
    session.rollback()

How is a unique constraint across three columns defined?

You need to add the constraint to the table, not the model. To do this using declarative:

class EventInvitation(db.Model):
    # ...
    __table_args__ = (
        db.UniqueConstraint(event_id, from_id, to_id),
    )

If the table has already been created in the database, you'll need to drop the table and run db.create_all() again, or use Alembic to alter the existing table with a migration.

Flask-SQLAlchemy create multi-column UniqueConstraint with multiple nullable columns

So, I've found out how to create an index with raw query. All you need is generate new migration with flask db migrate and modify both def upgrade() and 'def downgrade`:

def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    ...
    op.execute(f"CREATE UNIQUE INDEX unique_loc ON location "
               f"(longitude, latitude, COALESCE(address_1, '-1'), "
               f"COALESCE(address_2, '-1'), country_id, city_id);")
    ...
    # ### end Alembic commands ###

def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    ...
    op.drop_index('unique_loc', table_name='location')
    ...
    # ### end Alembic commands ###

SqlAlchemy: count of distinct over multiple columns

The exact query can be produced using the tuple_() construct:

session.query(
    func.count(distinct(tuple_(Hit.ip_address, Hit.user_agent)))).scalar()

---

Related Topics