擅长:python、mysql、java
<p>这是我根据@zzzeek的回答和最新评论提出的解决方案。我已经对它进行了单元测试,它似乎可以很好地处理回滚(会话在发出回滚后是干净的):</p>
<pre><code>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)
</code></pre>