Wiki

Clone wiki

sqlalchemy / UsageRecipes / ExpireRelationshipOnFKChange

ExpireRelationshipOnFKChange

The FAQ has for many years talked about how SQLAlchemy regards changes on foreign key attributes. When an operation like setting "foo_id" on an object "Foo" occurs, nothing happens to the associated relationships. The ORM is designed to work the other way; it will keep everything in sync as long as you deal with object relationships. If you want to modify column attributes directly, that's fine, but there's no "reverse" synchronization of this back to the relevant relationships.

Ticket #1939 also talks about the possibility of adding this feature; but it has caveats and uncertainties that still have us hesitant to add this feature. One primary issue is that SQLAlchemy, unlike many other ORMs, actually caches collections, rather than having them emit SQL every time. Which means to expire these collections on an attribute set operation means we have to actively go out and emit new SQL to find the new parent object, unless the relationship is a so-called "use_get many to one". The implications of these kinds of issues, combined with the fact that the SQLAlchemy ORM has a recommended pattern that avoids this issue entirely, keeps us hesitant to make this a supported feature.

However, SQLAlchemy has a comprehensive event API today, so we can produce a recipe version of this feature for simple cases, including that it can be customized to emit more or less SQL as needed. A full example is below.

from sqlalchemy.orm import object_session
from sqlalchemy import event

from sqlalchemy.orm.interfaces import MANYTOONE
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

def _expire_prop_on_col(cls, prop, colkey):
    @event.listens_for(getattr(cls, colkey), "set")
    def expire(target, value, oldvalue, initiator):
        sess = object_session(target)

        if sess is not None:
            # optional behavior #1 - expire the "User.articles"
            # collection on the existing "user" object
            if prop.back_populates and prop.key in target.__dict__:
                sess.expire(getattr(target, prop.key), [prop.back_populates])

            # behavior #2 - expire Article.user
            sess.expire(target, [prop.key])

            # optional behavior #3 - "trick" the ORM by actually
            # setting the value ahead of time, then emitting a load
            # for the attribute so that the *new* Article.user
            # is loaded.  Then, expire User.articles on *that*.
            # Other techniques here including looking in the identity
            # map for "value", if this is a simple many-to-one get.
            if prop.back_populates:
                target.__dict__[colkey] = value
                new = getattr(target, prop.key)
                if new is not None:
                    sess.expire(new, [prop.back_populates])



@event.listens_for(Base, "mapper_configured", propagate=True)
def mapper_configured(mapper, cls):
    for prop in mapper.relationships:
        if prop.direction is MANYTOONE:
            for col in prop.local_columns:
                colkey = mapper.get_property_by_column(col).key
                _expire_prop_on_col(cls, prop, colkey)

if __name__ == '__main__':
    from sqlalchemy.orm import object_session, Session, relationship
    from sqlalchemy import create_engine, Column, Integer, ForeignKey

    class User(Base):
        __tablename__ = 'user'
        id = Column(Integer, primary_key=True)

    class Article(Base):
        __tablename__ = 'article'
        id = Column(Integer, primary_key=True)
        content = Column(Integer)

        user_id = Column(Integer, ForeignKey(User.id))
        user = relationship("User", backref="articles")

    e = create_engine("sqlite://", echo=True)

    Base.metadata.create_all(e)

    DBSession = Session(e)

    user1 = User(id=23)
    user2 = User(id=42)
    DBSession.add(user2)
    DBSession.add(user1)

    article = Article()
    article.user_id = user1.id
    DBSession.add(article)

    DBSession.flush()
    assert article.user.id == 23
    assert article.user_id == 23

    article.user_id = 42

    DBSession.flush()
    assert article.user.id == 42
    assert article.user_id == 42

    DBSession.commit()
    assert article.user.id == 42
    assert article.user_id == 42

    assert article in user2.articles
    assert article not in user1.articles

    article.user_id = 23

    assert article not in user2.articles
    assert article in user1.articles

Updated