exploring usage of after_flush

sqlalchemy_after_flush.py

import sqlalchemy
from sqlalchemy import event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Table, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, scoped_session, relationship, backref
import datetime

Base = declarative_base()
engine = create_engine('sqlite://', echo=False)
Session = scoped_session(sessionmaker(bind=engine))

def associate(id1, id2, type1=Integer, type2=Integer):
    def get_table(id):
        return id.split('.')[0]

    def get_id(id):
        if '.' not in id:
            return id + '.id'
        return id

    def get_column(id):
        return get_table(id) + '_id'

    return Table(
        'link_%s_%s' % (get_table(id1), get_table(id2)), Base.metadata,
        Column(get_column(id1), type1, ForeignKey(get_id(id1))),
        Column(get_column(id2), type2, ForeignKey(get_id(id2)))
    )

class Author(Base):
    __tablename__ = 'author'

    id = Column(Integer, primary_key=True)
    name = Column(String(80))
    email = Column(String)

    def __repr__(self):
        return '%s(%r, %r)' % (self.__class__.__name__, self.id, self.name)

class Book(Base):
    __tablename__ = 'book'

    id = Column(Integer, primary_key=True)
    title = Column(String(80))
    year = Column(Integer, nullable=True)
    authors = relationship(Author,
                           secondary=Table(
                                'link_books_authors', Base.metadata,
                                Column('book_id', Integer,
                                       ForeignKey('book.id')),
                                Column('author_id', Integer,
                                       ForeignKey('author.id'))),
                           backref='books')

    def __repr__(self):
        return '%s(%r)' % (self.__class__.__name__, self.id)

class Chapter(Base):
    __tablename__ = 'chapter'

    id = Column(Integer, primary_key=True)
    book_id = Column(Integer, ForeignKey('book.id'))
    book = relationship(Book,
                        backref=backref('chapters', order_by=id))
    title = Column(String(80))

    def __repr__(self):
        return '%s(%r)' % (self.__class__.__name__, self.title)

@event.listens_for(Session, 'after_flush')
def after_flush(session, flush_context):
    print "-" * 40
    
    for state in flush_context.states:
        print state.object
        updated = False

        for prop in state.mapper.relationships:
            history = state.attrs[prop.key].load_history()
            added, unchanged, deleted = history

            if added:
                updated = True
                print "  added:"
                print "    {0!r:<12} {1!r}".format(prop.key, added)

            if deleted:
                updated = True
                print "  deleted:"
                print "    {0!r:<12} {1!r}".format(prop.key, deleted)

        if not updated:
            print " no relationships updated"

def test():
    Base.metadata.create_all(engine)

    # create a book
    book1 = Book(title='Book1')
    session = Session()
    session.add(book1)
    session.commit()

    # create another book and add authors to both
    john = Author(name='John Doe')
    jane = Author(name='Jane Doe')
    book1.authors = [john, jane]
    book2 = Book(title='Book2')
    book2.authors = [john]
    session.add(book2)
    session.commit()

    # add chapters to the books
    book1.chapters.append(Chapter(title='Intro to Book1'))
    chapter = Chapter(title='Intro to Book2', book=book2)
    session.commit()

    # move chapter
    chapter.book = book1
    session.commit()

if __name__ == '__main__':
    test()