Отслеживание изменений модели в SQLAlchemy

Я хочу регистрировать каждое действие, что будет сделано с некоторыми SQLAlchemy-Models.

Итак, у меня есть привязки after_insert, after_delete и before_update, где я сохраню предыдущее и текущее представление модели,

def keep_logs(cls):
    @event.listens_for(cls, 'after_delete')
    def after_delete_trigger(mapper, connection, target):
        pass

    @event.listens_for(cls, 'after_insert')
    def after_insert_trigger(mapper, connection, target):
        pass

    @event.listens_for(cls, 'before_update')
    def before_update_trigger(mapper, connection, target):
        prev = cls.query.filter_by(id=target.id).one()
        # comparing previous and current model


MODELS_TO_LOGGING = (
    User,
)
for cls in MODELS_TO_LOGGING:
    keep_logs(cls)

Но есть одна проблема: когда я пытаюсь найти модель в hook_update, SQLA возвращает измененную (грязную) версию. Как я могу получить предыдущую версию модели перед ее обновлением? Есть ли другой способ сохранить изменения модели?

Спасибо!

Ответ 1

SQLAlchemy отслеживает изменения каждого атрибута. Вам не нужно (и не следует) запрашивать экземпляр еще раз в событии. Кроме того, событие инициируется для любого экземпляра, который был изменен, даже если это изменение не изменит никаких данных. Зацикливайтесь на каждом столбце, проверяйте, был ли он изменен, и сохраняйте все новые значения.

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

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

        if not 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

Ответ 2

Если срок действия атрибута истек (какие сеансы выполняются по умолчанию при фиксации), старое значение недоступно, если оно не было загружено до изменения. Вы можете увидеть это при осмотре.

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=())

Для того, чтобы атрибуты оставались загруженными, вы не можете просрочить сущности, установив expire_on_commit в значение False в сеансе.

Ответ 3

У меня была похожая проблема, но я хотел иметь возможность отслеживать дельты, поскольку изменения вносятся в модели sqlalchemy, а не только в новые значения. Я написал это небольшое расширение к ответу Давидизма, чтобы сделать это вместе с немного лучшей обработкой before и after, так как они иногда являются списками или пустыми кортежами в других случаях:

  from sqlalchemy import inspect

  def get_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="[email protected]">'
      get_model_changes(user)
      >>> {}
      user.email = '[email protected]'
      get_model_changes(user)
      >>> {'email': ['[email protected]', '[email protected]']}
    """
    state = inspect(model)
    changes = {}
    for attr in state.attrs:
      hist = state.get_history(attr.key, True)

      if not hist.has_changes():
        continue

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

    return changes

  def has_model_changed(model):
    """
    Return True if there are any unsaved changes on the model.
    """
    return bool(get_model_changes(model))