You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在测试中禁用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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.11 08:05:24