Source code for examples.versioned_rows.versioned_rows
"""Illustrates a method to intercept changes on objects, turningan UPDATE statement on a single row into an INSERT statement, so that a newrow is inserted with the new data, keeping the old row intact."""fromsqlalchemyimportColumnfromsqlalchemyimportcreate_enginefromsqlalchemyimporteventfromsqlalchemyimportForeignKeyfromsqlalchemyimportIntegerfromsqlalchemyimportStringfromsqlalchemy.ext.declarativeimportdeclarative_basefromsqlalchemy.ormimportattributesfromsqlalchemy.ormimportbackreffromsqlalchemy.ormimportmake_transientfromsqlalchemy.ormimportrelationshipfromsqlalchemy.ormimportSessionfromsqlalchemy.ormimportsessionmakerclassVersioned:defnew_version(self,session):# make us transient (removes persistent# identity).make_transient(self)# set 'id' to None.# a new PK will be generated on INSERT.self.id=None@event.listens_for(Session,"before_flush")defbefore_flush(session,flush_context,instances):forinstanceinsession.dirty:ifnotisinstance(instance,Versioned):continueifnotsession.is_modified(instance):continueifnotattributes.instance_state(instance).has_identity:continue# make it transientinstance.new_version(session)# re-addsession.add(instance)Base=declarative_base()engine=create_engine("sqlite://",echo=True)Session=sessionmaker(engine)# example 1, simple versioningclassExample(Versioned,Base):__tablename__="example"id=Column(Integer,primary_key=True)data=Column(String)Base.metadata.create_all(engine)session=Session()e1=Example(data="e1")session.add(e1)session.commit()e1.data="e2"session.commit()assertsession.query(Example.id,Example.data).order_by(Example.id).all()==([(1,"e1"),(2,"e2")])# example 2, versioning with a parentclassParent(Base):__tablename__="parent"id=Column(Integer,primary_key=True)child_id=Column(Integer,ForeignKey("child.id"))child=relationship("Child",backref=backref("parent",uselist=False))classChild(Versioned,Base):__tablename__="child"id=Column(Integer,primary_key=True)data=Column(String)defnew_version(self,session):# expire parent's reference to ussession.expire(self.parent,["child"])# create new versionVersioned.new_version(self,session)# re-add ourselves to the parentself.parent.child=selfBase.metadata.create_all(engine)session=Session()p1=Parent(child=Child(data="c1"))session.add(p1)session.commit()p1.child.data="c2"session.commit()assertp1.child_id==2assertsession.query(Child.id,Child.data).order_by(Child.id).all()==([(1,"c1"),(2,"c2")])