05 分页、搜索、排序

本章目标：把文章列表接口升级为「专业版」：标准分页响应（带总数）、关键词搜索、多条件过滤、排序。这些是所有列表接口的标配能力。

1. 目标接口设计

GET /posts?page=1&size=10          分页
        &keyword=fastapi           标题/正文关键词搜索
        &author_id=1               按作者过滤
        &tag=python                按标签过滤
        &published=true            只看已发布
        &order_by=-created_at      排序（- 前缀表示降序）

响应带上分页元信息（前端需要总数来渲染页码）：

{
  "total": 42,
  "page": 1,
  "size": 10,
  "items": [ ...文章列表... ]
}

2. schemas.py：通用分页响应模型

用泛型定义一次，所有资源的列表接口都能复用：

from typing import Generic, TypeVar

T = TypeVar("T")


class Page(BaseModel, Generic[T]):
    """通用分页响应"""
    total: int
    page: int
    size: int
    items: list[T]

用法：Page[PostOut]、Page[UserOut]……Pydantic 完全支持泛型模型，/docs 里也能正确显示具体结构。

3. crud.py：升级 list_posts

from sqlalchemy import func, or_, select
from sqlalchemy.orm import selectinload

# 允许排序的字段白名单（防止外部传入任意字段名）
POST_ORDERABLE = {
    "id": models.Post.id,
    "created_at": models.Post.created_at,
    "updated_at": models.Post.updated_at,
    "title": models.Post.title,
}


def search_posts(
    db: Session,
    *,
    keyword: str | None = None,
    author_id: int | None = None,
    tag: str | None = None,
    published: bool | None = None,
    order_by: str = "-created_at",
    page: int = 1,
    size: int = 10,
) -> tuple[int, list[models.Post]]:
    """按条件搜索文章，返回 (总数, 当前页数据)"""

    stmt = select(models.Post)

    # ---- 动态拼接过滤条件：传了才加，没传不影响 ----
    if keyword:
        stmt = stmt.where(
            or_(
                models.Post.title.like(f"%{keyword}%"),
                models.Post.content.like(f"%{keyword}%"),
            )
        )
    if author_id is not None:
        stmt = stmt.where(models.Post.author_id == author_id)
    if published is not None:
        stmt = stmt.where(models.Post.published == published)
    if tag:
        # 多对多过滤：文章的标签中包含指定名字
        stmt = stmt.where(models.Post.tags.any(models.Tag.name == tag.lower()))

    # ---- 先算总数（基于同样的过滤条件，但不带分页） ----
    total = db.scalar(select(func.count()).select_from(stmt.subquery())) or 0

    # ---- 排序："-created_at" → created_at 降序 ----
    desc = order_by.startswith("-")
    field_name = order_by.lstrip("-")
    column = POST_ORDERABLE.get(field_name, models.Post.created_at)
    stmt = stmt.order_by(column.desc() if desc else column.asc())

    # ---- 分页 + 预加载 ----
    stmt = (
        stmt.options(selectinload(models.Post.author), selectinload(models.Post.tags))
        .offset((page - 1) * size)
        .limit(size)
    )
    return total, list(db.scalars(stmt).all())

值得学习的技巧：

动态拼接查询：select() 语句是不可变对象，每次 .where() 返回新语句，所以可以按条件逐步叠加——这是 ORM 相比手拼 SQL 字符串的巨大优势
relationship.any()：多对多/一对多的「存在性过滤」，生成 EXISTS 子查询，一行代码搞定「有 python 标签的文章」
排序白名单：绝不能把用户传的字符串直接当字段名用（getattr(Post, order_by) 是危险写法），白名单字典既安全又明确
先 count 后分页：总数必须基于「过滤后、分页前」的语句计算

*, 之后的参数都是关键字参数——参数多的函数强制关键字调用，可读性更好，防止传错位置。

4. routers/posts.py：升级列表接口

用一个类依赖收拢查询参数（第二部分第 7 章学过），替换原来的 list_posts：

from typing import Annotated
from fastapi import Depends, Query


class PostFilters:
    def __init__(
        self,
        page: Annotated[int, Query(ge=1)] = 1,
        size: Annotated[int, Query(ge=1, le=100)] = 10,
        keyword: Annotated[str | None, Query(max_length=50)] = None,
        author_id: int | None = None,
        tag: str | None = None,
        published: bool | None = None,
        order_by: Annotated[
            str,
            Query(description="排序字段，前缀 - 表示降序，如 -created_at"),
        ] = "-created_at",
    ):
        self.page = page
        self.size = size
        self.keyword = keyword
        self.author_id = author_id
        self.tag = tag
        self.published = published
        self.order_by = order_by


@router.get("", response_model=schemas.Page[schemas.PostOut])
def list_posts(filters: Annotated[PostFilters, Depends()], db: SessionDep):
    """文章列表：支持分页、搜索、过滤、排序"""
    total, items = crud.search_posts(
        db,
        keyword=filters.keyword,
        author_id=filters.author_id,
        tag=filters.tag,
        published=filters.published,
        order_by=filters.order_by,
        page=filters.page,
        size=filters.size,
    )
    return schemas.Page(
        total=total, page=filters.page, size=filters.size, items=items
    )

Depends() 括号里可以不写类名——参数类型注解已经是 PostFilters，FastAPI 会自动推断。

5. 测试

先多造几篇不同作者、不同标签、不同发布状态的文章，然后在 /docs 里试：

请求	预期
`GET /posts?size=2`	items 只有 2 条，total 是全部数量
`GET /posts?size=2&page=2`	第二页的 2 条
`GET /posts?keyword=fastapi`	标题或正文含 fastapi 的文章
`GET /posts?tag=python`	带 python 标签的文章
`GET /posts?author_id=1&published=true`	组合过滤
`GET /posts?order_by=title`	按标题升序
`GET /posts?order_by=-id`	按 id 降序
`GET /posts?size=9999`	422（size 上限 100）

再把 database.py 里的 echo=True 打开，观察每个请求实际生成的 SQL——count 一条、查询一条、selectinload 两条，共 4 条，与文章数量无关。

6. 举一反三

用同样的套路，你可以自己练习：

给 GET /users 也加上 Page[UserOut] 和 keyword 搜索（搜用户名/邮箱）
给文章加日期范围过滤：created_after、created_before 两个参数，stmt.where(Post.created_at >= created_after)

本章小结

分页响应用泛型模型 Page[T]，一次定义处处复用
动态拼接：条件传了才 .where()，语句可以逐步叠加
多对多过滤用 relationship.any()；排序字段必须走白名单
先基于过滤条件 count，再 offset/limit 取当页

第四部分完结！你已经拥有一个功能完整的博客 API。接下来给它加上真正的用户认证：05-进阶主题/01-用户认证JWT

1. 目标接口设计​

2. schemas.py：通用分页响应模型​

3. crud.py：升级 list_posts​

4. routers/posts.py：升级列表接口​

5. 测试​

6. 举一反三​

本章小结​