15 路由匹配中使用 ORM —— 从"返回空列表"到"真实的数据库查询"

前置阅读: 14-ORM建表、10-依赖注入
关键词: AsyncSession, Depends, await db.execute, 路由层, 数据访问
难度: ★★★★☆

场景导入：把第 04 章的占位端点，换成真实的 MySQL 查询

第 14 章已经把 Author 和 Book 模型落到代码，启动时 init_db 也完成了建表。但路由层至今仍返回内存字典或空列表——GET /books 永远返回 []，GET /books/42 永远返回一个拼凑的 {"id": 42, "title": "placeholder"}。

本期要把第 04 章登记的 GET /books 和 GET /books/{book_id} 改造为基于 SQLAlchemy ORM 的真实查询。处理函数不再关心连接怎么获得——它只声明"需要一个 AsyncSession"，框架通过 Depends(get_async_db) 按请求作用域注入。响应返回前，Session 自动关闭、连接归还到连接池。这是数据层章节和路由层章节的衔接点——从这一篇起，所有的 GET 请求都会真正打到 MySQL。

原理解析：依赖注入 + 异步 Session = 一行 await 串起所有

ORM 接入路由的核心是两层协同：依赖注入负责 Session 的创建和销毁，异步查询负责把 select 表达式翻译为 SQL 并拿到结果。

（图注：HTTP 请求进入路由后，Depends(get_async_db) 创建 AsyncSession 注入 handler，handler 通过 await db.execute 发起查询，响应生成后 async with 退出自动关闭 Session 并归还连接。）

整个流程中，几条关键约束决定了系统的可靠性：

一次请求一个 Session。 get_async_db 的 async def 生成器形态保证了 Session 在响应返回前不会跨请求复用——这是事务边界和身份映射正确性的前提。

业务代码不直接持有 Engine。 handler 拿到的是 AsyncSession，不是 Engine。这避免了连接池被业务代码直接操作，也避免了跨请求复用同一连接。

Session 的提交由业务代码控制。 本节只做查询不修改数据，Session 在 async with 退出时关闭；涉及新增、更新、删除时，业务代码应显式 await db.commit()，异常路径通过 await db.rollback() 回滚。这一语义将在第 17-19 章展开。

四层之间的依赖关系用一张图来看更清楚：

（图注：路由层只声明依赖并使用模型，依赖层管理异步 Session，模型层声明字段与关系，基础设施层提供异步 Engine 与 Base 元数据——四层各司其职。）

response_model 与 ORM 模型的衔接是一个容易被忽视的工程细节。路由层返回 list[Book]，FastAPI 序列化 ORM 实例时会触发 Book 类的属性读取。如果 handler 内部访问了 book.author.name，就会触发一次 lazy loading（额外的 SELECT），在响应序列化阶段产生 N+1 查询。本节示例只返回图书主表字段，不访问关联属性，规避这一陷阱。expire_on_commit=False 已让提交后的属性保持可读，不再触发隐式 SELECT——与第 13 章基础设施保持一致。

代码实现：GET /books 和 GET /books/{book_id}，接入 MySQL

下面的代码在 app/api/books.py 中把两个查询端点改造为基于 ORM 的异步查询。AsyncSession 通过 Depends(get_async_db) 注入，Book 模型来自第 14 章：

# app/api/books.py
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.db import get_async_db
from app.models.book import Book

router = APIRouter(prefix="/books", tags=["books"])
DBDep = Annotated[AsyncSession, Depends(get_async_db)]


@router.get("")
async def list_books(db: DBDep) -> list[Book]:
    """图书列表，基于 ORM 查询全表。"""
    result = await db.execute(select(Book).order_by(Book.id))
    return list(result.scalars().all())


@router.get("/{book_id}")
async def get_book(book_id: int, db: DBDep) -> Book | None:
    """图书详情，未命中时返回 None。"""
    result = await db.execute(select(Book).where(Book.id == book_id))
    return result.scalar_one_or_none()

几行代码背后藏着不少细节：

• DBDep = Annotated[AsyncSession, Depends(get_async_db)] 把类型和依赖来源绑定到别名，与第 10 章的写法一脉相承。handler 签名只写 db: DBDep，干净利落。

• select(Book).order_by(Book.id) 生成 SELECT * FROM books ORDER BY id。result.scalars().all() 把结果行映射为 Book 实例列表——注意 scalars() 去掉行对象外壳，.all() 一次性取出全部。

• select(Book).where(Book.id == book_id) 在主键上做等值查询，走聚簇索引 B+ 树定位。scalar_one_or_none() 未命中返回 None，多于一行抛异常——比静默取第一条更安全。

• await 是必须的。db.execute 返回的是协程，不加 await 你会拿到一个 coroutine 对象而不是 Result。

main.py 使用 lifespan 异步上下文管理器管理启动建表：

# main.py
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from fastapi import FastAPI
from app.api.books import router as books_router
from app.db import init_db


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncIterator[None]:
    await init_db()
    yield


app = FastAPI(title="图书管理 API", version="0.3.0", lifespan=lifespan)
app.include_router(books_router)

lifespan 是 FastAPI 0.93 之后引入的形态，替代了已废弃的 @app.on_event("startup")。在 ASGI 应用栈中位置更明确——它在应用开始接受请求之前完成建表，关闭期不需要额外清理（Engine 随进程退出释放）。

避坑指南

• select(Book) 不加 limit 会全表扫描。 图书数量增长到万级以上时，list_books 必须改为分页查询（第 16 章引入 limit/offset）。

• 非索引字段的等值查询是性能陷阱。 select(Book).where(Book.title == "...") 如果没有索引，数据库会全表扫描。生产环境应提前在迁移脚本中为高频查询列建立索引。

• 响应序列化可能触发 N+1。 如果 ORM 实例包含未加载的关联属性，response_model 在序列化阶段会触发 lazy loading。本节只返回主表字段，规避了这个问题；后续涉及关联数据时必须预加载。

• Session 关闭后别碰 ORM 对象。 访问已关闭 Session 中的对象属性会触发 DetachedInstanceError。业务代码应避免把 Session 中的实例传出 handler 存到全局变量或缓存中——跨请求只应传 book.id，下次请求重新查询。

面试 QA

Q1 [源码]: FastAPI 的 Depends(get_async_db) 与 SQLAlchemy 异步 Session 的生命周期是如何对齐的？

FastAPI 通过 async def 生成器依赖把 Session 的生命周期对齐到 HTTP 请求边界。请求进入路由时调用 get_async_db()，框架停在 yield 处把 AsyncSession 注入到 handler，handler 返回或抛异常后框架回到 yield 之后执行 async with 的退出关闭 Session。

源码层面，solve_dependencies 在 fastapi/dependencies/utils.py 中实现，通过 inspect.isasyncgenfunction 判定是否为异步生成器依赖——这是与同步 def yield 依赖的关键差异。SQLAlchemy 的 AsyncSession.close() 负责把连接归还到连接池。后续第 17-19 章的新增、更新、删除都依赖这一形态保证事务边界。

Q2 [项目]: FastAPI 集成异步 SQLAlchemy 后，如何组织 DAO 层与路由层的职责？

路由层只负责 HTTP 协议与依赖声明，DAO 层封装复杂查询与跨表关联，模型层声明字段与关系。DAO 层接受 AsyncSession 作为参数（从依赖注入中拿到），封装"按作者过滤""按状态分组"等复杂查询，返回 ORM 实例或值对象。路由层调用 DAO 而不是直接调用 await db.execute——从而隔离业务代码与 ORM 细节。

生产实践中，DAO 粒度应以"单一职责"为原则：避免一个巨型 BookDao 类，把不同查询拆为独立函数。Session 注入保持显式——DAO 函数签名接受 AsyncSession 而不是从全局变量获取，便于测试替换。性能埋点（慢查询日志、SQL 计数）也集中在 DAO 层。

小结

本期把第 04 章登记的 GET /books 与 GET /books/{book_id} 改造为基于异步 SQLAlchemy ORM 的真实查询。核心链路：Depends(get_async_db) 注入 AsyncSession → handler 调用 await db.execute(select(Book)) → result.scalars().all() 拿到 ORM 实例 → response_model 序列化为 JSON。这是从"同步占位"到"异步真实查询"的分水岭。

下一篇《16 数据操作之查询》将进一步扩展查询能力——引入 where 过滤、聚合统计和分页，并把复杂查询下沉到 DAO 层。