如何禁用SQLAlchemy查询缓存？解决测试场景下断言失败问题

问题背景 #

刚接触异步编程和SQLAlchemy，搭建测试套件时遇到以下问题：

按照文档实现了会话fixture：

@pytest.fixture( name="s", scope="function" )
async def fixture_s():
    connection  = await DB.engine.connect()
    transaction = await connection.begin()
    session     = Session( bind=connection )
    yield session
    session.close()
    await transaction.rollback()
    await connection.close()

Model类定义：

from sqlalchemy import String, func, select
from sqlalchemy.orm import Model, Mapped, mapped_column

class Thingy( Model ):
    __tablename__ = "thingies"
    id   : Mapped[ int ] = mapped_column( primary_key=True )
    name : Mapped[ str ] = mapped_column( String( 50 )     )

    @classmethod
    async def count( cls ):
        async with DB.Session() as s:
            res = await s.scalar( select( func.count( cls.id ) ) )
        return res

单独执行两个测试均通过：

async def test__foo( self, s ):
    assert 0 == await Thingy.count()

async def test__foo( self, s ):
    s.add( Thingy( name="foo" ) )
    assert 1 == await Thingy.count()

但合并后测试失败，第二个count调用始终返回0：

async def test__foo( self, s ):
    assert 0 == await Thingy.count()
    s.add( Thingy( name="foo" ) )
    assert 1 == await Thingy.count()

日志显示查询语句被缓存：

------------------------------ Captured log call -------------------------------
INFO     sqlalchemy.engine.Engine:base.py:2685 BEGIN (implicit)
INFO     sqlalchemy.engine.Engine:base.py:1844 SELECT count(thingies.id) AS count_1 
FROM thingies
INFO     sqlalchemy.engine.Engine:base.py:1844 [generated in 0.00016s] ()
INFO     sqlalchemy.engine.Engine:base.py:2688 ROLLBACK
INFO     sqlalchemy.engine.Engine:base.py:2685 BEGIN (implicit)
INFO     sqlalchemy.engine.Engine:base.py:1844 SELECT count(thingies.id) AS count_1 
FROM thingies
INFO     sqlalchemy.engine.Engine:base.py:1844 [cached since 0.004266s ago] ()
INFO     sqlalchemy.engine.Engine:base.py:2688 ROLLBACK

核心原因 #

日志中的cached指的是SQLAlchemy的语句缓存（复用生成的SQL语句），但这不是结果不更新的根本原因：

1. 测试会话s添加对象后未执行flush，对象仅存在于会话内存中，未写入数据库连接的事务上下文。
2. Thingy.count()每次新建独立会话，该会话与测试会话s虽共享同一个外部事务，但默认无法看到其他会话未提交/未刷新的修改。

解决方案 #

方法1：让count方法复用测试会话（推荐） #

修改Thingy.count()，允许传入外部会话，避免新建独立会话，直接读取测试会话的内存状态和事务上下文：

class Thingy( Model ):
    # ... 原有定义不变
    @classmethod
    async def count( cls, session=None ):
        if session:
            # 复用传入的会话，直接查询
            return await session.scalar( select( func.count( cls.id ) ) )
        # 保留原有逻辑，供非测试场景使用
        async with DB.Session() as s:
            return await s.scalar( select( func.count( cls.id ) ) )

修改测试代码，添加对象后执行flush，确保对象写入事务上下文：

async def test__foo( self, s ):
    assert 0 == await Thingy.count(session=s)
    s.add( Thingy( name="foo" ) )
    await s.flush()  # 将对象从会话内存同步到数据库连接的事务中
    assert 1 == await Thingy.count(session=s)

方法2：禁用语句缓存（仅解决表象，不推荐） #

若仅需禁用SQL语句缓存，可在查询时添加execution_options(cache_ok=False)：

@classmethod
async def count( cls ):
    async with DB.Session() as s:
        res = await s.scalar(
            select( func.count( cls.id ) ).execution_options(cache_ok=False)
        )
    return res

注意：此方法仅阻止SQL语句复用，无法解决跨会话的事务隔离问题，测试仍会失败，仅作知识补充。

方法3：让count会话加入现有事务 #

修改count方法，让新建的会话绑定到测试用的同一个连接事务：

@classmethod
async def count( cls, connection=None ):
    if connection:
        async with Session(bind=connection) as s:
            return await s.scalar( select( func.count( cls.id ) ) )
    async with DB.Session() as s:
        return await s.scalar( select( func.count( cls.id ) ) )

测试时传入测试会话的连接：

async def test__foo( self, s ):
    assert 0 == await Thingy.count(connection=s.connection())
    s.add( Thingy( name="foo" ) )
    await s.flush()
    assert 1 == await Thingy.count(connection=s.connection())

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