如何检查SQLAlchemy会话是否已修改

Question

我有一个SQLAlchemy的Session对象，想知道它是否被更改了。具体问题是：如果在此时我执行一个commit()或rollback()，对数据库的影响是否相同？原因是我想询问用户是否确认更改。但如果没有更改，我不想询问。当然，我可以自己监视所有在Session上执行的操作，并决定是否有修改，但由于我的程序结构，这将需要一些相当复杂的更改。如果SQLAlchemy已经提供了这个机会，我很乐意利用它。谢谢大家。

- Giovanni Mascellani

Answer 1

你正在寻找整个事务会话期间发生的实际刷新次数的净计数；虽然有一些线索可以判断是否发生了这种情况（称为“快照”），但此结构仅用于帮助回滚，并不是强引用。最直接的方法是跟踪“after_flush”事件，因为只有在调用flush并且flush找到要刷新的状态时才会发出此事件。

from sqlalchemy import event
import weakref
transactions_with_flushes = weakref.WeakSet()

@event.listens_for(Session, "after_flush")
def log_transaction(session, flush_context):
    for trans in session.transaction._iterate_parents():
        transactions_with_flushes.add(trans)

def session_has_pending_commit(session):
    return session.transaction in transactions_with_flushes

编辑：这里有一个更新版本，更简单易懂：

from sqlalchemy import event

@event.listens_for(Session, "after_flush")
def log_transaction(session, flush_context):
    session.info['has_flushed'] = True

def session_has_pending_commit(session):
    return session.info.get('has_flushed', False)

- zzzeek

---

我在一年半后注意到了这个答案...谢谢，这正是我所需要的。我通过添加另一个检查来增强它：只有在session.new、session.deleted或session.dirty中有内容时，事务才会被添加到transactions_with_flushes中。此外，session.dirty中的对象首先要使用session.is_modified进行测试。最终的实现在我的这个项目中：https://github.com/giomasce/liturgy/blob/master/database.py#L19。它似乎可以工作，尽管我没有进行疯狂的测试。 - Giovanni Mascellani

_iterate_parents 已从 SQLAlchemy 中移除。现在必须手动进行父级迭代。请参见上面的链接以查看我的实现，它似乎可以工作。 - Giovanni Mascellani

这个答案无论如何都很奇怪。现在Session有一个.Info字典。在after_flush中将"flushed=True"放入其中，在after_commit中将其删除。我不确定为什么要将所有的SessionTransaction对象放入一个字典中。 - zzzeek

@zzzeek这个在回滚操作中表现良好吗？或者我们也需要钩入after_事件来实现回滚操作？能否请你用info字典方法更新你的答案？ - rdrey

@rdrey，提交/回滚与flush()无关。上述方法还会告诉您是否已经刷新了受到rollback()影响的数据。 - zzzeek

Answer 2

这是基于@zzzeek的答案和更新评论的解决方案。我已经进行了单元测试，它似乎与回滚良好地配合使用（在发出回滚后，会话将变为干净状态）：

from sqlalchemy import event
from sqlalchemy.orm import Session


@event.listens_for(Session, "after_flush")
def log_flush(session, flush_context):
    session.info['flushed'] = True


@event.listens_for(Session, "after_commit")
@event.listens_for(Session, "after_rollback")
def reset_flushed(session):
    if 'flushed' in session.info:
        del session.info['flushed']


def has_uncommitted_changes(session):
    return any(session.new) or any(session.deleted) \
        or any([x for x in session.dirty if session.is_modified(x)]) \
        or session.info.get('flushed', False)

- rdrey

---

这是最完整的答案。 - ntc2

Answer 3

该会话具有dirty属性

session.dirty

当前检测到更改的持久化对象（每次调用该属性时都会创建此集合）

sqlalchemy.orm.session.Session.dirty

- Yoriz

---

这并不能解决我的问题：可能没有脏对象，但如果某些更改已经刷新到数据库中，我仍然想知道。 - Giovanni Mascellani

Answer 4

会话拥有一个私有的_is_clean()成员，如果没有需要刷新到数据库的内容，则返回true。然而，它是私有的，这意味着它可能不适合外部使用。我个人不建议使用它，因为任何错误显然都会导致用户数据丢失。

- Kylotan

---

除了它是私有的这一事实外，它似乎与Yoriz的答案存在相同的问题：我想检测已经刷新但未提交的更改。 - Giovanni Mascellani

嗯，好的。我没有其他建议，除了在SQLAlchemy组（https://groups.google.com/forum/?fromgroups#!forum/sqlalchemy）上发布问题，如果你得到答案，请在这里回复。 - Kylotan