SQLAlchemy重执行语句示例异常：'FrozenResult'无'scalars'属性排查

我想通过SQLAlchemy实现缓存功能，调研后认为do_orm_execute是最优方案，参考官方文档实现了简易版本，但遇到无法解决的错误。

我的代码：

def do_orm_execute(orm_execute_state):
    if not orm_execute_state.is_select:
        return
    
    global cache
    cache_key = f"{orm_execute_state.statement}-{orm_execute_state.parameters}"   
     
    if cache_key in cache:
        frozen_result = cache[cache_key]
    else:
        frozen_result = orm_execute_state.invoke_statement().freeze()
        cache[cache_key] = frozen_result    

    return loading.merge_frozen_result(
        orm_execute_state.session,
        orm_execute_state.statement,
        frozen_result,
        load=False,
    )

报错信息：

File "/Users/X/repos/X/X/venv/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2782, in one_or_none
return self._iter().one_or_none() # type: ignore
File "/Users/X/repos/X/X/venv/lib/python3.10/site-packages/sqlalchemy/orm/query.py", line 2863, in _iter
result = cast("Result[_T]", result).scalars()
AttributeError: 'FrozenResult' object has no attribute 'scalars'

极简复现代码：

def do_orm_execute(self, orm_execute_state):
    if orm_execute_state.is_select:           
        frozen_result = orm_execute_state.invoke_statement().freeze()
    
        return loading.merge_frozen_result(
            orm_execute_state.session,
            orm_execute_state.statement,
            frozen_result,
            load=False,
        )

问题根源 #

do_orm_execute事件回调要求返回ORM执行结果对象（如Result实例），但你直接返回了merge_frozen_result的输出——该方法返回的FrozenResult是底层数据结构，不具备Query API（如one_or_none()）所需的scalars()方法，因此触发属性错误。

修复方案 #

需将FrozenResult包装成兼容上层API的ORMResult实例，或利用orm_execute_state.result()方法构造符合要求的结果对象。调整后的代码示例：

方案1：手动包装为ORMResult

from sqlalchemy.orm import ORMResult

def do_orm_execute(orm_execute_state):
    if not orm_execute_state.is_select:
        return
    
    global cache
    cache_key = f"{orm_execute_state.statement}-{orm_execute_state.parameters}"   
     
    if cache_key in cache:
        frozen_result = cache[cache_key]
    else:
        result = orm_execute_state.invoke_statement()
        frozen_result = result.freeze()
        cache[cache_key] = frozen_result    

    merged = loading.merge_frozen_result(
        orm_execute_state.session,
        orm_execute_state.statement,
        frozen_result,
        load=False,
    )
    # 将合并后的结果包装为ORMResult返回
    return ORMResult(merged, orm_execute_state.session, orm_execute_state.statement)

方案2：利用orm_execute_state.result()构造结果

def do_orm_execute(orm_execute_state):
    if not orm_execute_state.is_select:
        return
    
    global cache
    cache_key = f"{orm_execute_state.statement}-{orm_execute_state.parameters}"   
     
    if cache_key in cache:
        frozen_result = cache[cache_key]
    else:
        result = orm_execute_state.invoke_statement()
        frozen_result = result.freeze()
        cache[cache_key] = frozen_result    

    merged = loading.merge_frozen_result(
        orm_execute_state.session,
        orm_execute_state.statement,
        frozen_result,
        load=False,
    )
    # 通过result方法生成兼容的返回对象
    return orm_execute_state.result(merged)

关键说明 #

• merge_frozen_result仅负责将冻结结果与当前会话关联，返回的FrozenResult不支持Query API的方法调用。
• ORMResult或orm_execute_state.result()会将底层结果包装成上层API期望的对象，提供scalars()、one_or_none()等必要方法。

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