FastAPI系列-16-数据操作之查询
meytao meytao 0 评论 Fastapi 1 阅读

FastAPI系列-16-数据操作之查询

_

16 数据操作之查询 —— 从"查全部"到"按需过滤、分页、聚合"

前置阅读: 15-路由匹配中使用ORM、05-参数分类
关键词: filter, like, in_, 聚合, 分页, DAO, async def
难度: ★★★★☆

场景导入：真实的 /books 接口不可能只查全表

第 15 章把 GET /books 接入了 MySQL，但它只会做一件事——SELECT * FROM books ORDER BY id。真实的图书列表接口需要按作者筛选、按状态过滤、按标题关键词搜索，还要返回总数并分页。

如果把这些判断全部写在路由函数里，HTTP 参数解析、SQL 条件拼装和分页逻辑会缠成一团。本章在已有的 AsyncSession 和 Book 模型之上引入 DAO（Data Access Object）层——路由负责参数校验和调用，DAO 负责构造 select 查询。重点是三类能力：where 条件组合、聚合函数统计、OFFSET 分页，同时为百万级数据量的键集分页留好演进空间。

原理解析：DAO 是路由和 ORM 之间的隔离带

DAO 的角色可以用一句话概括：路由通过 Depends(get_async_db) 拿到 Session，解析查询参数后调用 DAO 函数；DAO 接收 Session 并返回 ORM 对象或统计值。路由不感知 SQL，DAO 不碰 HTTP——各管各的。

SQLAlchemy 2.x 的查询构造遵循"惰性组合 → 一次执行"的模式。select(Book) 创建一个 Select 对象，然后链式追加 .where(...)、.order_by(...)、.offset(...)、.limit(...)——每一步都是纯 Python 表达式组合，不发起任何数据库 I/O。真正的 I/O 边界只有一处：await db.execute(stmt)。

flowchart LR A["select(Book)"] --> B["where 过滤"] B --> C["order_by 稳定排序"] C --> D["offset / limit 分页"] D --> E["await db.execute"] E --> F["scalars().all() 列表"] E --> G["scalar_one() 聚合"]

（图注：查询语句先惰性组合，到 await db.execute 才发出 SQL，结果再按实体列表或聚合值分别读取。）

过滤条件用 where 和 and_ 组合。Book.author_id == author_id 产生等值比较，Book.title.ilike(f"%{keyword}%") 做不区分大小写的模糊匹配，Book.status.in_(statuses) 翻译为 IN (...)。多个可选参数收集后用 and_(*conditions) 一次性加入 where——条件为空时 where 不追加，等价于查全表。

聚合由 sqlalchemy.func 提供。func.count(Book.id) 统计行数，func.sum、func.avg、func.max、func.min 分别做求和、平均和边界值。聚合查询的关键是只投影需要的列——select(func.count(Book.id)) 而不是 select(Book) 再取长度，避免把全部行加载到内存再计数。

分页方面，OFFSET 分页用 .offset((page - 1) * size).limit(size)，优点是可直接跳页，但深页时数据库仍要扫描并丢弃前面的行——第 100 页意味着前 99 页的所有行都被扫过。百万级表应迁移到键集分页（keyset pagination）：按 id 降序时，客户端传上一页最后的 after_id，DAO 追加 Book.id < after_id 再加 .limit(size)，数据库从主键索引锚点继续读取，延迟不随页码线性增长。

sequenceDiagram participant R as Router participant D as DAO participant S as AsyncSession participant M as MySQL R->>S: Depends(get_async_db) R->>D: await list_books(db, 条件) D->>D: select + where + limit D->>S: await db.execute(stmt) S->>M: 异步发送 SQL M-->>S: Result S-->>D: scalars().all() D-->>R: list[Book]

（图注：路由只编排参数与调用，DAO 在 await db.execute(stmt) 处跨过异步 I/O 边界。）

代码实现：过滤、分页、聚合，收进 DAO

下面的 book_dao.py 把图书过滤、统计和状态集合查询封装为三个 async def 函数。keyword 参数中的 LIKE 通配符（%、_、\）先转义，避免用户输入意外变成 SQL 通配符：

# app/dao/book_dao.py
from typing import Optional

from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.book import Book, BookStatus


async def list_books(
    db: AsyncSession,
    *,
    author_id: Optional[int] = None,
    status: Optional[BookStatus] = None,
    keyword: Optional[str] = None,
    offset: int = 0,
    limit: int = 20,
) -> list[Book]:
    """按条件过滤与分页，返回图书列表。"""
    stmt = select(Book)
    conditions = []
    if author_id is not None:
        conditions.append(Book.author_id == author_id)
    if status is not None:
        conditions.append(Book.status == status)
    if keyword:
        safe = keyword.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        conditions.append(Book.title.ilike(f"%{safe}%", escape="\\"))
    if conditions:
        stmt = stmt.where(and_(*conditions))
    stmt = stmt.order_by(Book.id.desc()).offset(offset).limit(limit)
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def count_books(
    db: AsyncSession,
    *,
    author_id: Optional[int] = None,
    status: Optional[BookStatus] = None,
) -> int:
    """按条件统计图书总数，用于分页元数据。"""
    stmt = select(func.count(Book.id))
    conditions = []
    if author_id is not None:
        conditions.append(Book.author_id == author_id)
    if status is not None:
        conditions.append(Book.status == status)
    if conditions:
        stmt = stmt.where(and_(*conditions))
    result = await db.execute(stmt)
    return int(result.scalar_one())


async def list_books_by_statuses(
    db: AsyncSession,
    statuses: list[BookStatus],
) -> list[Book]:
    """按状态集合过滤，演示 in_ 的典型用法。空集合直接返回空列表。"""
    if not statuses:
        return []
    stmt = select(Book).where(Book.status.in_(statuses))
    result = await db.execute(stmt)
    return list(result.scalars().all())

几个设计决策值得留意：

  • 关键字参数加 * 强制调用方使用命名传参。list_books(db, author_id=1, limit=10) 比 list_books(db, 1, None, None, 0, 10) 可读性好一个数量级，参数多了以后尤其明显。

  • None 哨兵区分"未提供"和"显式传 None"。如果调用方不传 status，DAO 不加这个过滤条件；如果传了 status=None，同样不加——不把 None 当作"查状态为 NULL 的行"。

  • LIKE 转义不能省。keyword.replace("%", "\\%").replace("_", "\\_") 配上 ilike(..., escape="\\")，让 % 和 _ 被当作字面字符而不是通配符。不转义的话用户搜"50% off"会匹配到所有标题。

  • 空 statuses 直接返回空列表，避免生成 WHERE status IN () 这种非法 SQL。

路由层负责把 HTTP 查询参数翻译为 DAO 调用参数：

# app/api/books.py
from typing import Annotated, Optional

from fastapi import APIRouter, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

from app.dao import book_dao
from app.db import get_async_db
from app.models.book import Book, BookStatus

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


@router.get("")
async def list_books(
    db: DBDep,
    author_id: Optional[int] = Query(default=None, gt=0),
    status: Optional[BookStatus] = Query(default=None),
    keyword: Optional[str] = Query(default=None, max_length=64),
    page: int = Query(default=1, ge=1),
    size: int = Query(default=20, ge=1, le=100),
) -> list[Book]:
    """图书列表，支持按作者、状态、关键字过滤，支持分页。"""
    return await book_dao.list_books(
        db,
        author_id=author_id,
        status=status,
        keyword=keyword,
        offset=(page - 1) * size,
        limit=size,
    )

handler 只做参数校验（gt=0、ge=1、max_length=64）和偏移量计算，具体的 SQL 拼装全部在 DAO 层完成。这种职责分离让单元测试可以在不启动 FastAPI 的情况下直接调用 book_dao.list_books(test_session, author_id=1) 验证查询逻辑。

避坑指南

  • ilike 的大小写行为由 MySQL collation 决定。utf8mb4_unicode_ci 默认不区分大小写，utf8mb4_bin 区分。上线前应按业务需求选择排序规则，而不是靠猜。

  • LIKE '%keyword%' 通常无法充分利用 B+ 树索引。前缀模糊（LIKE 'keyword%'）能走索引，全模糊只能扫全表。数据量大时评估全文索引（FULLTEXT）或搜索服务，用 EXPLAIN 验证查询计划。

  • 深页 OFFSET 是性能杀手。OFFSET 10000 LIMIT 20 意味着 MySQL 扫描并丢弃 10000 行然后取 20 行。列表增长后，键集分页（WHERE id < after_id ORDER BY id DESC LIMIT 20）的性能不受页码影响。

  • 列表和总数的统计应尽量保持事务一致性。如果先查列表再查总数，中间的并发写入可能导致总数和实际列表长度不匹配。同一次请求内使用同一个 AsyncSession，MySQL 的 MVCC 会保证读取到的是一致快照。

面试 QA

Q1 [原理]: select 风格与旧式 Query 风格有什么区别？

SQLAlchemy 2.x 推荐 select(Book).where(...) 返回可组合的 Select 对象，之后通过 await db.execute(stmt) 执行。旧式 Query 把构造和结果方法混在同一个对象上（如 session.query(Book).filter(...).all()）。两种写法最终由 SQLAlchemy 编译为相同 SQL，差异主要在 2.x API 的统一性、类型提示支持以及与 Core 层语句的组合能力。select 本身就是同步构造操作——不涉及 I/O，真正的异步边界是 await db.execute。

Q2 [项目]: 数据量增长到百万级，如何提升列表查询性能？

三步走。第一步，为常用过滤组合建立联合索引，用 EXPLAIN 验证命中——比如 (author_id, status, id) 索引能让"按作者 + 状态过滤 + 按 id 排序"走完整的索引覆盖。第二步，深页从 OFFSET 迁移到键集分页——DAO 通过 after_id 参数维持 API 兼容，路由层只多收一个可选参数。第三步，用"是否存在下一页"（has_next = len(items) > size，多取一条）替代昂贵的 COUNT 总数查询——大部分用户根本不会翻到最后一页。

小结

本章把第 15 章的简单全表查询扩展为完整的异步 DAO 查询层：可选条件用 and_ 动态组合，集合过滤用 in_，统计用 func.count，列表用稳定排序和 OFFSET 分页。book_dao.py 和路由都统一使用 AsyncSession、Depends(get_async_db) 和 await db.execute(stmt)——为后续新增、更新、删除章节的事务写入打好基础。

下一篇《17 数据操作之新增》将进入 db.add、await db.commit() 和 await db.refresh() 的新增流程，把第 06 章占位的 POST /books 改造为真实的数据写入端点。

文章标签
#fastapi#后端
许可协议
本文采用 CC BY-NC-SA 许可协议，转载请注明出处。
FastAPI系列-17-数据操作之新增 2026-07-03
FastAPI系列-15-路由匹配中使用ORM 2026-06-29

评论区