24. FastAPI 使用 SQLAlchemy 2 Async 查询关系数据

查询分成“选什么”“怎么 await”和“怎么加载关系”

select(User) 选择用户，.where(...) 过滤，.order_by(...) 排序，.limit(...) 限制数量。通过 AsyncSession 执行时必须 await db.execute() 或 await db.scalars()。

关系加载策略决定 user.profile、user.posts、user.roles 是否已经填充。异步环境不能依赖属性访问时隐式发 SQL，因此预加载比同步代码中更重要。

沿用第 20、21 篇项目，先用第 22 篇准备少量测试数据。脚本示例使用主库 SessionLocal；FastAPI 示例使用第 20 篇的 get_db，配置了从库时会走 ReadSessionLocal。

基本异步查询与结果形状

保存为 query_basics.py：

import asyncio

from sqlalchemy import func, select

from database import SessionLocal, master_engine
from models import Role, User


async def query_basics() -> None:
    async with SessionLocal() as db:
        user = await db.get(User, 1)  # 主键查找；不存在为 None

        users = (
            await db.scalars(
                select(User)
                .where(User.active.is_(True))
                .order_by(User.id)
                .limit(20)
            )
        ).all()

        result = await db.execute(select(User.id, User.name).limit(20))
        rows = result.mappings().all()

        total = await db.scalar(select(func.count()).select_from(User))

        with_posts = (
            await db.scalars(select(User).where(User.posts.any()))
        ).all()

        readers = (
            await db.scalars(
                select(User)
                .where(User.roles.any(Role.name == "reader"))
                .limit(20)
            )
        ).all()

        print("指定用户", user)
        print("总数", total, "当前页", len(users))
        print("列查询", [dict(row) for row in rows])
        print("有文章的用户", len(with_posts), "读者", len(readers))


async def main() -> None:
    try:
        await query_basics()
    finally:
        await master_engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

await db.scalars() 返回 ScalarResult，之后的 .all() 是读取已返回结果，不需要再次 await。await db.execute() 返回 Result，mappings() 按字段名访问。需要恰好一条可用 scalar_one()，允许零条可用 scalar_one_or_none()，出现多条时会报错。

常见错误是漏掉 await：

# 错误：得到的只是可等待对象，不是查询结果
result = db.execute(select(User))

# 正确
result = await db.execute(select(User))
users = result.scalars().all()

三种关系的完整详情接口

以下作为这个练习项目的 main.py。get_db 来自第 20 篇的 deps.py：

from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import joinedload, selectinload

from deps import get_db
from models import User

app = FastAPI()


class ORMOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)


class ProfileOut(ORMOut):
    bio: str


class PostOut(ORMOut):
    id: int
    title: str


class RoleOut(ORMOut):
    id: int
    name: str


class UserOut(ORMOut):
    id: int
    name: str
    profile: ProfileOut | None
    posts: list[PostOut]
    roles: list[RoleOut]


def user_statement():
    return select(User).options(
        joinedload(User.profile),
        selectinload(User.posts),
        selectinload(User.roles),
    )


@app.get("/users/{user_id}", response_model=UserOut)
async def detail(
    user_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
):
    user = await db.scalar(user_statement().where(User.id == user_id))
    if user is None:
        raise HTTPException(404, "用户不存在")

    # 关系已预加载，Pydantic 读取属性时不会再发异步 I/O
    return UserOut.model_validate(user)


@app.get("/users", response_model=list[UserOut])
async def users(
    db: Annotated[AsyncSession, Depends(get_db)],
    after_id: Annotated[int, Query(ge=0)] = 0,
    size: Annotated[int, Query(ge=1, le=100)] = 20,
):
    rows = (
        await db.scalars(
            user_statement()
            .where(User.id > after_id)
            .order_by(User.id)
            .limit(size)
        )
    ).all()
    return [UserOut.model_validate(row) for row in rows]

启动：

uv run uvicorn main:app --reload

请求真实用户 ID，响应同时包含一份资料、多篇文章和多个角色。输出模型不反向包含 Post.user、Role.users，避免无限递归和无意公开额外数据。

异步延迟加载、N+1 与 JOIN 重复

同步 SQLAlchemy 默认 lazy load 可能在每次访问关系时补发 SQL，形成 N+1；异步代码中，这种隐式 I/O 还可能直接抛出 MissingGreenlet。

推荐在查询语句中明确加载：

• joinedload(User.profile)：常用于一对一等多对一、标量关系。
• selectinload(User.posts)、selectinload(User.roles)：常用于集合关系，避免主查询 JOIN 后产生大量重复行。
• await db.refresh(user, attribute_names=["posts"])：在已有对象上显式刷新指定关系。

集合特别大时仍应单独分页查询文章，不能把一个用户的一百万篇文章全部嵌入响应。

使用 joinedload(User.posts) 加载集合时，同一用户会出现在多行 SQL 结果中，SQLAlchemy 2 通常要求对结果调用 .unique()：

from sqlalchemy import select
from sqlalchemy.orm import joinedload

from database import SessionLocal
from models import User


async def load_with_joined_posts():
    async with SessionLocal() as db:
        result = await db.scalars(
            select(User).options(joinedload(User.posts)).limit(20)
        )
        return result.unique().all()

注意顺序：先 await db.scalars(...) 得到 ScalarResult，再调用同步的 .unique().all()。

手动 JOIN 后再分页要格外小心：分页的可能是联结行，而不是用户数。父资源分页优先先确定用户集合，再加载关系。any()、has() 可表达关联存在性，适合“有某类文章的用户”这类过滤。

百万条异步查询与优化

API 只返回有限页；离线导出使用主键游标分批，不对全表调用 .all()。异步驱动的流式结果仍受 MySQL 驱动缓冲行为影响，yield_per 或 stream() 不等于可以忽略内存、事务时长和关系加载策略。

导出时先投影必要列，减少宽字段传输。索引应匹配过滤和排序，例如批次任务可评估 (batch_tag, id) 联合索引。使用 MySQL EXPLAIN 验证执行计划，不要假设每个字段加索引都会更快。

参数化 SQL 可以避免值注入；动态排序列仍需白名单。需要并发查询多个页面时使用独立 AsyncSession，并限制并发数，不能共享同一个 Session 调用 asyncio.gather()。

练习：分别用未预加载、selectinload 查询 20 个用户，观察 SQL 数量以及直接访问未加载关系时的错误；再按 batch_tag 统计用户、资料、文章和角色关联数量，验证第 22、23 篇的结果。