如何让SQLAlchemy从数据库重新查询而非复用会话缓存对象?
我使用SQLAlchemy 2.0.25和PostgreSQL 15.3编写项目仓储层测试,现有insert_or_update(name: str)函数通过insert().on_conflict_do_update()操作数据库,配合returning()返回操作后的对象。测试流程中,修改该对象后再次调用insert_or_update(),随后执行select()期望得到全新对象以验证更新结果,但发现insert_or_update()和select()返回的是同一内存地址的对象,修改其中一个会同步改变另一个。这是因为SQLAlchemy的会话通过标识映射(identity map)复用对象提升效率,以下是测试场景下绕过缓存、从数据库获取全新数据的可行方法:
方法1:创建新会话查询
执行完insert_or_update并提交后,关闭当前会话,用新会话执行查询。新会话无旧对象缓存,会从数据库重新加载数据,返回完全独立的新实例:
# 原会话完成更新操作 original_session.commit() original_session.close() # 初始化新会话查询 new_session = Session(bind=engine) fresh_obj = new_session.get(YourModel, target_id)
方法2:标记对象为过期
调用session.expire(obj)将指定对象标记为过期,下次访问该对象属性时,SQLAlchemy会自动从数据库重新拉取最新数据:
# 执行更新后标记对象过期 session.expire(updated_obj) # 再次查询时会重新加载数据库数据 fresh_obj = session.get(YourModel, updated_obj.id)
方法3:强制刷新对象属性
若无需新实例,仅需验证属性是否更新,可使用session.refresh(obj)直接从数据库同步最新属性值到现有对象:
# 执行更新后刷新对象 session.refresh(updated_obj) # 此时updated_obj的属性已同步数据库最新状态
方法4:查询时强制覆盖缓存
使用execution_options(populate_existing=True)执行查询,强制SQLAlchemy从数据库加载数据并覆盖会话中已有的对象实例(仍为同一内存对象,但属性会更新):
fresh_obj = session.query(YourModel)\ .filter_by(id=target_id)\ .execution_options(populate_existing=True)\ .first()
如果测试必须对比不同内存地址的对象(比如验证仅指定字段被修改),方法1是最可靠的选择,新会话返回的实例完全独立,不会受旧对象修改影响。
内容的提问来源于stack exchange,提问作者William

