Home > Net >  SQLAlchemy unique constrain by field
SQLAlchemy unique constrain by field

Time:06-15

I have UniqueConstraint on field, but it wont allow me to add multiple entries (two is max!)

from sqlalchemy import Column, Integer, String, Boolean, UniqueConstraint

class Cart(SqlAlchemyBase):
    __tablename__ = 'cart'
    __table_args__ = (UniqueConstraint('is_latest'), {})
    sid = Column(Integer, primary_key=True)
    is_latest = Column(Boolean, index=True, nullable=False)
    name = Column(String)

I would like to support more entries, so that one name can have two variants:

name=foo, is_latest=True
name=foo, is_latest=False
name=bar, is_latest=True
name=bar, is_latest=False

but then reject any subsequent attempt to write name=foo (or bar) and is_latest=True

CodePudding user response:

What you are trying to achieve here is a type 2 slowly changing dimension, this is a topic that has been discussed extensively and I encourage you to look it up.

When I look at your table you seem to use sid as a surrogate key, but I fail to see what is the natural key and what will be updated as time goes.

Anyway, there are several ways to achieve SCD type 2 result without the need to worry about your check, but the the simplest in my mind is to keep on adding records with your natural key and when querying, select only the one with highest surrogate key (autoincrementing integer), no need for current uniqueness here as only the latest value is fetched.

There are examples for versioning rows in SQLAlchemy docs, but since website come and go, I'll put a simplified draft of the above approach here.

class VersionedItem(Versioned, Base):
    id = Column(Integer, primary_key=True)  # surrogate key
    sku = Column(String, index=True)  # natural key
    price = Column(Integer)  # the value that changes with time

@event.listens_for(Session, "before_flush")
def before_flush(session, flush_context, instances):
    for instance in session.dirty:
        if not (
            isinstance(instance, VersionedItem)
            and session.is_modified(instance)
            and attributes.instance_state(instance).has_identity
        ):
            continue
        make_transient(instance)  # remove db identity from instance
        instance.id = None  # remove surrogate key
        session.add(instance)  # insert instance as new record

CodePudding user response:

Looks like a Partial Unique Index can be used:

class Cart(SqlAlchemyBase):
    __tablename__ = 'cart'
    id = Column(Integer, primary_key=True)
    cart_id = Column(Integer)
    is_latest = Column(Boolean, default=False)
    name = Column(String)

    __table_args__ = (
        Index('only_one_latest_cart', name, is_latest,
              unique=True,
              postgresql_where=(is_latest)),
    )
name=foo, is_latest = True
name=foo, is_latest = False
name=bar, is_latest = False
name=bar, is_latest = False

And when adding another name=foo, is_latest = True

psycopg2.errors.UniqueViolation: duplicate key value violates unique constraint "only_one_latest_cart"
DETAIL:  Key (name, is_latest)=(foo, t) already exists.
  • Related