Wiki

Clone wiki

sqlalchemy / UsageRecipes / UniqueConstraintCollection

UniqueConstraintCollection

Normally, when an item with a particular primary key is marked as deleted, and replaced by another object with the same key, SQLAlchemy sees that both of these objects have the same key, and turns the operation into a single UPDATE.

The issue here arises when either the objects are unique on a non-primary key UNIQUE constraint, or when the UPDATE behavior is not wanted, and a DELETE before INSERT is definitely needed. The SQLAlchemy unit of work, while it handles an extremely wide variety of complex dependencies, currently falls short at just this one. So we can use an event to emit the DELETE ahead of time as items are removed from a collection:

from sqlalchemy import Column, Integer, String, create_engine, ForeignKey
from sqlalchemy.orm import relationship, Session, object_session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import event

Base= declarative_base()

class Parent(Base):
    __tablename__ = 'parent'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", cascade="all, delete-orphan")

class Child(Base):
    __tablename__ = 'child'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parent.id'))
    name = Column(String, unique=True)

@event.listens_for(Parent.children, "remove")
def rem(state, item, initiator):
    sess = object_session(item)

    # ensure we have a session
    assert sess is not None

    # ensure the item is marked deleted.  the cascade
    # rule may have done so already but not always.
    sess.delete(item)

    # flush *just this one item*.  This is a special
    # feature of flush, not for general use.
    sess.flush([item])

e = create_engine("sqlite://", echo=True)
Base.metadata.create_all(e)

s = Session(e)

p1 = Parent(children=[
    Child(name='c1'),
    Child(name='c2')
])
s.add(p1)
s.commit()

p1.children = [
    Child(name='c2'),
    Child(name='c3')
]
s.commit()

Updated