SQLAlchemy ORM 2.0 - create audits (Insert/ Update/ Delete) MySQL

[emburse-manikamidha]

I am trying to create audits (get columns with values before and after Update) Update statement per ORM 2.0 style :
Example :

engine = create_engine("sqlite:///sales.db", echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)
session = Session()

class User12(Base):
    __tablename__ = "user12"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

Base.metadata.create_all(engine)

session.execute(insert(User12), [{ "name": "old_name" , "email": "[email protected]"}])
session.commit()
session.close() 

After this, if I follow Tracking model changes in SQLAlchemy to track updates, it works, provided I update a certain way. Example :
user = session.execute(select(User12).filter_by(name="old_name")).scalar_one()

This way of updating data can be tracked : user.email = '[email protected]'
But, not the ORM 2.0 style update - https://docs.sqlalchemy.org/en/20/orm/queryguide/dml.html#orm-queryguide-update-delete-where

stmt = (
       update(User12)
       .where(User12.name == "old_name")
       .values(email='[email protected]')
   )
session.execute(stmt)
session.commit() 

Can any one please share a working example ? FYI, I am trying to capture audits in a dictionary form - {col1: [old value, new value]}
I have tried get_history method as mentioned at Tracking model changes in SQLAlchemy but it only works with a specific kind of way of updating not with orm 2.0 style.

Thanks

[zzzeek]

note there are two different ways to do UPDATES with SQLAlchemy ORM.

1. use the unit of work. This is session.add(), session.flush(), and it's what you see when dealing with "model changes", which is remains the default way of using the Session, including in SQLAlchemy 2.0. Nothing has changed. Detailed documentation on these events is at https://docs.sqlalchemy.org/en/20/orm/session_events.html#persistence-events https://docs.sqlalchemy.org/en/20/orm/session_events.html#object-lifecycle-events etc.

2. for performance reasons, one might want to skip using the unit of work which works on a row-by-row basis and use bulk operations. The bulk operations are described at the documentation you link, at https://docs.sqlalchemy.org/en/20/orm/queryguide/dml.html. these techniques are completely optional and are only provided to allow a SQL-statement based way of working. If you would like to continue using the state tracking events from section one, dont use the bulk methods.

From there, we can look at the bulk method in question. This form:

stmt = (
       update(User12)
       .where(User12.name == "old_name")
       .values(email='[email protected]')
   )
session.execute(stmt)
session.commit() 

is a command that one uses to emit this exact SQL to the database:

UPDATE table SET email='new_email' WHERE name='old_name'

That's it! if you log into your database and type that command, it will run the UPDATE. What are the "old" values in the table? Well when you emit an UPDATE, they're gone! SQLAlchemy has no additional powers in this area, it just emits commands and gets results, just as you might do on your database's command line tool. if you want to know what "name" was previously, you would need to SELECT it first before updating the table. If this is PostgreSQL, there may be ways to use CTEs to mix these two operations into a single statement that uses RETURNING, and I would ask on PostgreSQL related boards how to get the old value of a row while UPDATEing it at the same time.

short answer though, nothing has changed. Use the same unit of work techniques everyone has used for decades.

[emburse-manikamidha]

@zzzeek Thanks a ton for your quick reply. Just to be sure, based on what I need to do (create audits by capturing old and new values) , I should update/ insert/delete the following way :
Example for Update :

user = session.execute(select(User12).filter_by(name="old_name")).scalar_one()
user.email = '[email protected]'

def get_instance_changes(user):
    state = inspect(user)
    changes = {}

    for attr in state.attrs:
        hist = attr.load_history()

        if hist.has_changes():
            old_val = hist.deleted[0] if hist.deleted else None
            new_val = hist.added[0] if hist.added else None

            changes[attr.key] = [old_val, new_val]

    return changes

print(get_instance_changes(user))

[zzzeek]

more or less yes, you'd want to flush and/or commit your session to persist those changes

[emburse-manikamidha]

Thank you. Really appreciate your help.