sqlalchemy_after_flush_issue.py

sqlalchemy_after_flush_issue.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))

class Book(Base):
    __tablename__ = 'book'

    id = Column(Integer, primary_key=True)
    title = Column(String(80))

    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,
                        # causes traceback:
                        # active_history=True,
                        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)

    print
    print "setup books"
    # create a book
    book1 = Book(title='Book1')
    book2 = Book(title='Book2')
    session = Session()
    session.add(book1)
    session.add(book2)

    # add chapters to the books
    #  update relationship from the 'many' side:
    chapter1 = Chapter(title='Chapter1')
    book1.chapters.append(chapter1)
    #  update relationship from the 'one' side:
    chapter2 = Chapter(title='Chapter2', book=book2)
    session.commit()

    print
    print "move chapter from the 'many' side (GOOD)"
    #  CORRECT BEHAVIOR:
    book1.chapters = []
    book2.chapters.append(chapter1)
    session.commit()

    print
    print "move chapter from the 'one' side (BAD)"
    #  PROBLEM: book1.chapters and book2.chapters DO NOT appear in load_history()
    chapter1.book = book1
    session.commit()

    print
    print "move chapter from the 'one' side (GOOD)"
    #  CORRECT BEHAVIOR: book1.chapters and book2.chapters appear in load_history()
    #  however, getting this behavior requires accessing the relationships first.
    #  this is not necessary when creating (see above), only when updating.
    book1.chapters
    book2.chapters
    chapter1.book = book2
    session.commit()

if __name__ == '__main__':
    test()