python - 如何检查 SQLAlchemy 会话是否脏

Question

我有一个 SQLAlchemySession对象，想知道它是否脏。我想（隐喻地）问的确切问题Session是：“如果此时我发出 acommit()或 a rollback()，对数据库的影响是否相同？”。

理由是：我想问用户他是否想要确认更改。但如果没有变化，我不想问任何事情。当然，我可能会监控自己执行的所有操作Session并决定是否有修改，但由于我的程序结构，这需要一些相当复杂的更改。如果 SQLAlchemy 已经提供了这个机会，我很乐意利用它。

谢谢大家。

Answer 1

您正在寻找在整个会话事务范围内进行的实际刷新的净计数；虽然有一些关于这是否发生的线索（称为“快照”），但这种结构只是为了帮助回滚而不是强引用。最直接的方法是跟踪“after_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)

Answer 2

会话具有脏属性

session.dirty

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

sqlalchemy.orm.session.Session.dirty

Answer 3

这是我基于@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)

Answer 4

会话有一个私有_is_clean()成员，如果没有任何东西要刷新到数据库，它似乎返回 true。但是，它是私有的这一事实可能意味着它不适合外部使用。我不会亲自推荐这个，因为这里的任何错误显然都可能导致您的用户丢失数据。