SQLAlchemy未提交会话查询对象问题及bauble/ghini异常排查

一、Bauble/Ghini中Rollback无效的异常原因 #

从你的测试结果能明显看到，在Bauble环境中执行session.flush()后，即使调用session.rollback()，数据库中的数据依然保留——这完全符合你提到的"自定义Session执行flush时会触发commit"的描述。

标准SQLAlchemy中，flush()只是把会话内的变更同步到数据库的未提交事务中，此时rollback()可以完全撤销这些变更；但如果Bauble的自定义Session重写了flush方法，在内部调用了commit()，那么flush()执行后事务就已经被持久化到数据库，后续的rollback()自然无效（因为没有活跃事务可以回滚了）。

排查与验证方法 #

你可以直接查看Bauble代码库中bauble/db.py的Session类实现：

• 检查是否重写了flush方法，比如类似：

  class Session(sqlalchemy.orm.Session):
      def flush(self, *args, **kwargs):
          super().flush(*args, **kwargs)
          self.commit()  # 这里会直接提交事务
  

• 或者是否通过SQLAlchemy事件监听（比如after_flush）绑定了自动提交逻辑：

  from sqlalchemy import event
  @event.listens_for(Session, 'after_flush')
  def after_flush(session, context):
      session.commit()
  

临时 Workaround #

如果暂时无法修改Bauble的Session实现，又需要避免flush导致的自动提交，可以尝试使用嵌套事务：

session = Session()
session.query(Location).delete()
session.commit()

u1 = Location(code=u'mario')
session.add(u1)

# 开启嵌套事务
with session.begin_nested():
    session.flush()  # 此时flush只会提交嵌套事务，外层事务仍未提交
    u2 = session.query(Location).filter_by(code=u'mario').one()
    print(u1, u2, u1==u2)

# 回滚外层事务，会撤销嵌套事务的变更
session.rollback()
u3 = session.query(Location).filter_by(code=u'mario').first()
print(u3)  # 此时应该输出None

二、不执行flush查询会话内对象的方法 #

标准SQLAlchemy中，session.query()默认会优先查询数据库（即使会话内有未flush的对象），但我们可以手动从会话的内部集合中筛选对象，无需触发flush。

方法1：手动筛选会话集合 #

SQLAlchemy的Session维护了几个核心集合：

• session.new：尚未flush的新对象
• session.dirty：已修改但未flush的对象
• session.identity_map：已flush/提交的对象（按主键索引）

你可以写一个辅助函数，先从这些集合中查找，再 fallback 到数据库查询：

def query_session_first(session, model, **filters):
    # 先查identity_map（已同步到数据库的对象）
    for obj in session.identity_map.values():
        if isinstance(obj, model):
            match = all(getattr(obj, k) == v for k, v in filters.items())
            if match:
                return obj
    # 再查session.new（未flush的新对象）
    for obj in session.new:
        if isinstance(obj, model):
            match = all(getattr(obj, k) == v for k, v in filters.items())
            if match:
                return obj
    # 最后查数据库
    return session.query(model).filter_by(**filters).first()

使用方式：

session.add(u1)
u2 = query_session_first(session, Location, code=u'mario')
print(u2)  # 直接输出mario，无需flush

方法2：利用SQLAlchemy的Query扩展（进阶） #

如果你希望保持session.query()的语法风格，可以自定义一个Query类，重写其查询逻辑，优先从会话集合中查找：

from sqlalchemy.orm import Query

class SessionFirstQuery(Query):
    def first(self):
        # 先尝试从会话中查找
        model = self._entities[0].entity_zero.class_
        filters = {}
        # 解析查询条件（简单示例，复杂条件需调整）
        for criterion in self._criterion:
            if hasattr(criterion, 'left') and hasattr(criterion, 'right'):
                col_name = criterion.left.name
                filters[col_name] = criterion.right.value
        result = query_session_first(self.session, model, **filters)
        if result is not None:
            return result
        # 否则执行原查询逻辑
        return super().first()

# 绑定到Session（如果是Bauble的自定义Session，需确认是否支持替换query_cls）
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(query_cls=SessionFirstQuery)

这样你就可以直接用session.query(Location).filter_by(code=u'mario').first()拿到会话内的对象，无需flush。

内容的提问来源于stack exchange，提问作者mariotomo