Tracking Model Changes In Sqlalchemy

I want to log every action what will be done with some SQLAlchemy-Models. So, I have a after_insert, after_delete and before_update hooks, where I will save previous and current re

Solution 1:

SQLAlchemy tracks the changes to each attribute. You don't need to (and shouldn't) query the instance again in the event. Additionally, the event is triggered for any instance that has been modified, even if that modification will not change any data. Loop over each column, checking if it has been modified, and store any new values.

@event.listens_for(cls, 'before_update')defbefore_update(mapper, connection, target):
    state = db.inspect(target)
    changes = {}

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

        ifnot hist.has_changes():
            continue# hist.deleted holds old value# hist.added holds new value
        changes[attr.key] = hist.added

    # now changes map keys to new values

Solution 2:

I had a similar problem but wanted to be able to keep track of the deltas as changes are made to sqlalchemy models instead of just the new values. I wrote this slight extension to davidism's answer to do that along with slightly better handling of before and after, since they are lists sometimes or empty tuples other times:

from sqlalchemy import inspect

  defget_model_changes(model):
    """
    Return a dictionary containing changes made to the model since it was 
    fetched from the database.

    The dictionary is of the form {'property_name': [old_value, new_value]}

    Example:
      user = get_user_by_id(420)
      >>> '<User id=402 email="business_email@gmail.com">'
      get_model_changes(user)
      >>> {}
      user.email = 'new_email@who-dis.biz'
      get_model_changes(user)
      >>> {'email': ['business_email@gmail.com', 'new_email@who-dis.biz']}
    """
    state = inspect(model)
    changes = {}
    for attr in state.attrs:
      hist = state.get_history(attr.key, True)

      ifnot hist.has_changes():
        continue

      old_value = hist.deleted[0] if hist.deleted elseNone
      new_value = hist.added[0] if hist.added elseNone
      changes[attr.key] = [old_value, new_value]

    return changes

  defhas_model_changed(model):
    """
    Return True if there are any unsaved changes on the model.
    """returnbool(get_model_changes(model))

Solution 3:

If an attribute is expired (which sessions do by default on commit) the old value is not available unless it was loaded before being changed. You can see this with the inspection.

state = inspect(entity)
session.commit()
state.attrs.my_attribute.history  # History(added=None, unchanged=None, deleted=None)# Load history manually
state.attrs.my_attribute.load_history()
state.attrs.my_attribute.history  # History(added=(), unchanged=['my_value'], deleted=())

In order for attributes to stay loaded you can not expire entities by settings expire_on_commit to False on the session.