如何在测试中禁用SQLAlchemy-Continuum自动填充事务表?
问题
我在基于Flask+MSSQL的项目中使用sqlalchemy-continuum对指定数据库表做版本控制。端到端测试时,我会启动Flask应用上下文,为Flask全局g对象填充用户信息,该信息由sqlalchemy-continuum的UserPlugin处理。现在我需要通过flask-fixtures库从手写JSON文件加载精确的数据库状态,且需要完全控制数据库内的数据。但当JSON中包含版本化表的数据时,sqlalchemy-continuum会自动填充transaction表,导致我无法控制该表中存储的用户信息。
我尝试在加载JSON fixture时执行以下操作:
# 判断JSON中是否包含我提供的transaction表数据 if fixture["model"] == "transaction": special_transaction = fixture ... # 加载所有普通表数据的代码,之后执行: if special_transaction is not None: statement = text( "insert into [transaction] " "(issued_at, remote_addr, user_id) " "values (:remote_addr, :issued_at, :user_id)" ) session = <获取session的代码> for record in fixture["records"]: session.execute(statement, record) session.commit()
目前该操作无法生效,因为sqlalchemy-continuum已经自动填充了transaction表,导致插入自定义数据时出现冲突。请问如何关闭该自动填充行为,以便在测试中完全控制数据库状态?
解决方法
1. 临时禁用sqlalchemy-continuum的事务自动创建
在加载fixture的代码块中,通过设置session的continuum_disable属性,临时关闭continuum的自动事务跟踪:
# 获取session后立即禁用continuum session = <获取session的代码> session.continuum_disable = True # 执行flask-fixtures的加载逻辑(包括版本化表和transaction表数据) # ... # 加载完成后恢复continuum功能(如果后续测试需要) session.continuum_disable = False session.commit()
2. 临时移除UserPlugin插件
如果需要避免UserPlugin自动填充用户信息,可以在加载fixture期间临时移除该插件:
from sqlalchemy_continuum import versioning_manager from sqlalchemy_continuum.plugins import UserPlugin # 保存原插件列表,临时移除UserPlugin original_plugins = versioning_manager.plugins.copy() versioning_manager.plugins = [p for p in versioning_manager.plugins if not isinstance(p, UserPlugin)] # 执行fixture加载操作 # ... # 加载完成后恢复原插件配置 versioning_manager.plugins = original_plugins
3. 用原生数据库连接绕过ORM钩子
如果是通过原生SQL插入transaction表,可以直接使用数据库连接而非ORM session,跳过continuum的事件监听:
session = <获取session的代码> # 获取原生数据库连接 conn = session.connection() if special_transaction is not None: statement = text( "insert into [transaction] " "(issued_at, remote_addr, user_id) " "values (:issued_at, :remote_addr, :user_id)" ) for record in special_transaction["records"]: conn.execute(statement, record) session.commit()
4. 清空自动生成数据后插入自定义内容
如果前面的方法不适用,可以先清空transaction表中continuum自动生成的记录,再插入自定义数据:
if special_transaction is not None: session = <获取session的代码> # 清空自动生成的transaction记录 session.execute(text("DELETE FROM [transaction]")) # 插入自定义数据 statement = text( "insert into [transaction] " "(issued_at, remote_addr, user_id) " "values (:issued_at, :remote_addr, :user_id)" ) for record in special_transaction["records"]: session.execute(statement, record) session.commit()
内容的提问来源于stack exchange,提问作者user19837762
相关产品推荐
相关产品推荐

