SqlAlchemy AsyncIO ORM查询时报AsyncSession无query属性如何解决？

报错根因 #

SQLAlchemy 异步ORM的AsyncSession实例不支持同步场景下的query方法，所有查询操作需要先通过sqlalchemy.select构造查询语句，再调用异步执行方法处理。

基础依赖导入 #

首先需要新增select构造器的导入：

from sqlalchemy import select

常用查询场景实现 #

1. 查询全表数据（对应缓存预加载场景） #

将原来的SESSION.query(TableClass).all()替换为以下写法：

CACHE = {}
async def _load_all():
    global CACHE
    try:
        # 构造查询语句
        stmt = select(TableClass)
        # 异步执行查询
        result = await SESSION.execute(stmt)
        # 提取所有ORM对象结果
        curr = result.scalars().all()
        CACHE = {i.id: i.alias for i in curr}
    finally:
        # 异步场景下scoped_session使用完成后必须移除，归还连接到连接池
        await SESSION.remove()

LOOP.run_until_complete(_load_all())

2. 按主键查询单条数据（对应原query(TableClass).get(x)场景） #

有两种简洁实现方式：

方式1：直接调用AsyncSession内置的get方法（推荐，和原同步写法逻辑最接近）

async def get_table_obj(pk: int):
    # 参数1为ORM模型类，参数2为主键值，查询不到返回None
    return await SESSION.get(TableClass, pk)

方式2：手动构造select语句查询

async def get_table_obj(pk: int):
    stmt = select(TableClass).filter_by(id=pk)
    result = await SESSION.execute(stmt)
    # scalar()返回第一条匹配结果，无匹配返回None
    return result.scalar()

原有代码注意事项 #

• 你的表模型类需要继承BASE才能被元数据自动创建表，需修改定义为class TableClass(BASE):
• 全局复用scoped_session时必须在每次操作完成后调用await SESSION.remove()，避免连接泄漏
• 如果是在Web服务（如FastAPI）中使用，建议每个请求创建独立的Session实例，请求结束后自动关闭，避免协程安全问题

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