在 SQLAlchemy 中跟踪模型变化

Question

我想记录一些 SQLAlchemy 模型将要完成的每个操作。

所以，我有一个 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)

但是有一个问题：当我试图在 before_update 挂钩中查找模型时，SQLA returns 修改（脏）版本。如何在更新之前获取模型的先前版本？是否有其他方法来保持模型更改？

谢谢！

Answer 1

SQLAlchemy 跟踪每个属性的变化。您不需要（也不应该）在事件中再次查询实例。此外，任何已修改的实例都会触发该事件，即使该修改不会更改任何数据。遍历每一列，检查它是否已被修改，并存储任何新值。

@event.listens_for(cls, 'before_update')
def before_update(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

Answer 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 使实体过期。

Answer 3

我遇到了类似的问题，但希望能够在对 sqlalchemy 模型进行更改时跟踪增量，而不仅仅是新值。我写了这个对 davidism's 答案的轻微扩展，同时稍微更好地处理 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="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)

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

在 SQLAlchemy 中跟踪模型变化

Tracking model changes in SQLAlchemy

python

sqlalchemy

unit-of-work