-
- from sqlalchemy import create_engine, Column
- from sqlalchemy.types import Integer
- from sqlalchemy.orm import scoped_session, sessionmaker
- from sqlalchemy.ext.declarative import declarative_base
- from sqlalchemy.event import listens_for
-
-
- Base = declarative_base()
- Session = scoped_session(sessionmaker())
-
-
- class MyModel(Base):
- __tablename__ = "MyTable"
- priid = Column(Integer(), primary_key=True)
- secid = Column(Integer())
-
-
- @listens_for(MyModel, 'after_insert')
- def mymodel_after_insert(mapper, connection, target):
- mytable = MyModel.__table__
- priid = target.priid
- statement = (mytable.update()
- .where(mytable.c.priid == priid)
- .values(secid=priid))
- connection.execute(statement)
-
-
- if __name__ == '__main__':
- engine = create_engine('sqlite://', echo=True)
- Base.metadata.create_all(bind=engine)
- Session.configure(bind=engine)
-
- # Add 10 rows
- for i in xrange(10):
- Session.add(MyModel())
-
- Session.commit()
- Session.remove()
-
- # Check the rows
- rows = Session.query(MyModel).all()
- for row in rows:
- assert row.secid == row.priid
-
- print("Checked %s rows." % len(rows))
-