import datetime
from sqlalchemy import Column, ForeignKey, Integer, DateTime, Table, create_engine, func, String, event
from sqlalchemy.orm import sessionmaker, relationship, configure_mappers, declarative_base
from sqlalchemy_history import make_versioned, version_class
make_versioned(user_cls=None)
Base = declarative_base()
book_author_table = Table(
'book_author',
Base.metadata,
Column('book_id', Integer, ForeignKey('book.id'), primary_key=True, nullable=False),
Column('author_id', Integer, ForeignKey('author.id'), primary_key=True, nullable=False),
Column('created_date', DateTime, nullable=False, server_default=func.current_timestamp(), default=datetime.datetime.utcnow),
)
class Book(Base):
__tablename__ = 'book'
__versioned__ = {}
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(20), nullable=False)
authors = relationship('Author', secondary=book_author_table, back_populates='books')
class Author(Base):
__tablename__ = 'author'
__versioned__ = {}
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(20), nullable=False)
books = relationship('Book', secondary=book_author_table, back_populates='authors')
configure_mappers()
engine = create_engine('sqlite:///temp.db', echo=False)
# Create all tables
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
# Create a session
session = sessionmaker()
session.configure(bind=engine)
db = session()
lotr = Book(name='Lord of the rings')
tolkien = Author(name='JRR Tolkien', books=[lotr])
@event.listens_for(session, 'after_flush_postexec')
def after_flush_postexec(session, flush_context):
if tolkien.name != "yoyoyoyoyo":
tolkien.name = "yoyoyoyoyo"
new_user = Author(name='JRR Tolkien', books=[lotr])
db.add(new_user)
db.add(lotr)
db.add(tolkien)
db.commit()
print(db.query(version_class(Author)).all())
# >> [AuthorVersion(id=1, transaction_id=1, operation_type=1), AuthorVersion(id=2, transaction_id=1, operation_type=0)]
# Author with id =1 should have operation_type as 1 but it is 0
tolkien.name = "yoyoyoyoyo1"
db.add(lotr)
db.add(tolkien)
db.commit()
print(db.query(version_class(Author)).all())
# >> [AuthorVersion(id=1, transaction_id=1, operation_type=1), AuthorVersion(id=2, transaction_id=1, operation_type=0), AuthorVersion(id=1, transaction_id=2, operation_type=1), AuthorVersion(id=3, transaction_id=2, operation_type=0)]
# first record should have operation type as 0
When adding
after_flush_postexechook for modifying some values in DB after flush, the operation type is messed upCode that reproduces this issue