SQLAlchemy异步Session替代同步query查询方法的技术咨询

之前使用同步Session时，无论采用命令式映射还是声明式映射，都可通过query方法查询映射到表的模型。但改用AsyncSession后，调用.query()会报错：'AsyncSession' object has no attribute 'query'。当前依赖版本：SQLAlchemy1.4.41、asyncpg0.27.0。

原代码示例 #

模型与表定义 #

import sqlalchemy
from sqlalchemy import Table, Column, String, Integer, UUID
from dataclasses import dataclass
import uuid

mapper_registry = sqlalchemy.orm.registry()

@dataclass(unsafe_hash=True)
class MissionStatus:
    uuid: UUID
    name: str
    order: int

missions_status = Table(
    "mission_status",
    mapper_registry.metadata,
    Column(
        "uuid",
        UUID(as_uuid=True),
        primary_key=True,
        index=True,
        unique=True,
        default=uuid.uuid4,
    ),
    Column("name", String, nullable=False),
    Column("order", Integer, nullable=False),
)

命令式映射初始化 #

mapper_registry.map_imperatively(
    MissionStatus,
    missions_status,
)

原同步查询语句 #

my_missionstatus = session.query(MissionStatus).filter(MissionStatus.order == 1).first()

解决办法 #

SQLAlchemy 1.4的AsyncSession不支持同步API中的.query()方法，需改用核心查询构造器select配合session.execute()，同时所有异步操作需加await关键字。

修改后的异步查询代码 #

1. 导入select构造器：

from sqlalchemy import select

2. 执行异步查询（两种写法任选）：

# 写法1：分步执行
result = await session.execute(select(MissionStatus).filter(MissionStatus.order == 1))
my_missionstatus = result.scalar_one_or_none()  # 等价于原first()，无结果返回None

# 写法2：简化链式调用
my_missionstatus = await session.scalar(select(MissionStatus).filter(MissionStatus.order == 1))

关键说明 #

• 异步Session的所有数据库操作必须通过await调用
• 核心查询语句（select/insert/update等）是SQLAlchemy 1.4+同步/异步API统一的查询方式
• 命令式映射的模型无需额外修改，直接用select(MissionStatus)即可查询

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