I’m working on a project using SQLAlchemy to interact with a database. Currently, I’m implementing an auditing functionality that requires accessing the previous values of objects before updates are made to the database.

I’ve tried using the before_flush event to capture the previous values before updates are made, but I’m getting in values_before and values_after both the new commited values.

I’ve also tried @event.listens_for(SomeClass, 'before_update') with the same results.

Here’s my current implementation:

def register_audit_events(db):
    @event.listens_for(db.session, 'before_flush')
    def before_flush(session, flush_context, instances):
        for obj in session.dirty:
            if isinstance(obj, (StoreModel, ItemModel, TagModel)):
                # Here I try to store de previous values
                obj._previous_values = {c.name: getattr(obj, c.name) for c in obj.__table__.columns}

    @event.listens_for(db.session, 'after_flush')
    def after_flush(session, flush_context):
        user_id = session.info.get('user_id')
        
        for obj in session.new:
            if isinstance(obj, (StoreModel, ItemModel, TagModel)):
                operation = 'CREATE'
                values = json.dumps({c.name: getattr(obj, c.name)
                                    for c in obj.__table__.columns})
                log_audit_event(obj, operation, user_id, None, values)

        for obj in session.dirty:
            if isinstance(obj, (StoreModel, ItemModel, TagModel)):
                # Auditoría del evento de actualización
                operation = 'UPDATE'
                values_before = json.dumps(obj._previous_values)  # I use the supposed previous values 
                values_after = json.dumps(
                    {c.name: getattr(obj, c.name) for c in obj.__table__.columns})
                log_audit_event(obj, operation, user_id,
                                values_before, values_after)

        for obj in session.deleted:
            if isinstance(obj, (StoreModel, ItemModel, TagModel)):
                operation = 'DELETE'
                values = json.dumps({c.name: getattr(obj, c.name)
                                    for c in obj.__table__.columns})
                log_audit_event(obj, operation, user_id, values, None)