Release: 1.3.24 legacy version | Release Date: March 30, 2021

SQLAlchemy 1.3 Documentation

Contents | Index | Download as ZIP file

Source code for examples.versioned_rows.versioned_rows

"""Illustrates a method to intercept changes on objects, turning
an UPDATE statement on a single row into an INSERT statement, so that a new
row is inserted with the new data, keeping the old row intact.

from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import event
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import attributes
from sqlalchemy.orm import backref
from sqlalchemy.orm import make_transient
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm import sessionmaker

class Versioned(object):
    def new_version(self, session):
        # make us transient (removes persistent
        # identity).

        # set 'id' to None.
        # a new PK will be generated on INSERT. = None

@event.listens_for(Session, "before_flush")
def before_flush(session, flush_context, instances):
    for instance in session.dirty:
        if not isinstance(instance, Versioned):
        if not session.is_modified(instance, passive=True):

        if not attributes.instance_state(instance).has_identity:

        # make it transient
        # re-add

Base = declarative_base()

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

Session = sessionmaker(engine)

# example 1, simple versioning

class Example(Versioned, Base):
    __tablename__ = "example"
    id = Column(Integer, primary_key=True)
    data = Column(String)


session = Session()
e1 = Example(data="e1")
session.commit() = "e2"

assert session.query(, == (
    [(1, "e1"), (2, "e2")]

# example 2, versioning with a parent

class Parent(Base):
    __tablename__ = "parent"
    id = Column(Integer, primary_key=True)
    child_id = Column(Integer, ForeignKey(""))
    child = relationship("Child", backref=backref("parent", uselist=False))

class Child(Versioned, Base):
    __tablename__ = "child"

    id = Column(Integer, primary_key=True)
    data = Column(String)

    def new_version(self, session):
        # expire parent's reference to us
        session.expire(self.parent, ["child"])

        # create new version
        Versioned.new_version(self, session)

        # re-add ourselves to the parent
        self.parent.child = self


session = Session()

p1 = Parent(child=Child(data="c1"))
session.commit() = "c2"

assert p1.child_id == 2
assert session.query(, == (
    [(1, "c1"), (2, "c2")]