11-分页与过滤
Python 3.11+
本章讲解 FastAPI 中的分页与过滤实现，涵盖偏移分页、游标分页、查询过滤、排序及最佳实践。
第一部分：分页基础（L1）
1.1 为什么需要分页
实际场景：数据库中有 100 万条记录，一次性返回会导致内存溢出、响应超时、网络带宽浪费。
问题：如何高效地分批返回数据？
┌─────────────────────────────────────────────────────────────┐
│ 不分页的代价 │
├──────────────┬──────────────────────────────────────────────┤
│ 数据库层 │ SELECT * FROM items → 全表扫描，内存占用高 │
│ 应用层 │ 100 万行 ORM 对象 → Python 内存溢出风险 │
│ 序列化层 │ JSON 序列化 100 万条 → CPU 密集，耗时数秒 │
│ 网络层 │ 响应体数百 MB → 超时、客户端解析失败 │
│ 用户体验 │ 页面白屏等待 → 用户流失 │
└──────────────┴──────────────────────────────────────────────┘分页的核心价值：
- 性能：减少数据库扫描范围和内存占用
- 用户体验：快速响应，渐进式加载
- 资源控制：限制单次请求的 CPU、网络、内存消耗
1.2 偏移分页（Offset/Limit）
最常见的分页方式，通过 OFFSET 跳过前 N 条记录，LIMIT 限制返回数量。
python
from fastapi import FastAPI, Query
app: FastAPI = FastAPI()
@app.get("/items")
def get_items(
page: int = Query(default=1, ge=1, description="页码，从 1 开始"),
size: int = Query(default=20, ge=1, le=100, description="每页数量"),
) -> dict[str, int | list[str]]:
offset: int = (page - 1) * size
# SQL: SELECT * FROM items LIMIT {size} OFFSET {offset}
# 模拟数据
all_items: list[str] = [f"item_{i}" for i in range(1000)]
page_items: list[str] = all_items[offset : offset + size]
return {
"page": page,
"size": size,
"total": len(all_items),
"items": page_items,
}偏移分页的工作原理：
请求: GET /items?page=3&size=10
SQL 执行流程:
┌───────────────────────────────────────────┐
│ 1. 扫描所有行（全表或索引扫描） │
│ 2. 跳过前 20 行 (OFFSET = (3-1)*10) │
│ 3. 返回接下来的 10 行 (LIMIT 10) │
│ │
│ [1][2]...[20] │ [21][22]...[30] │ [31]... │
│ ↑ 跳过 ↑ 返回 ↑ 不关心 │
└───────────────────────────────────────────┘1.3 游标分页（Cursor/After）
基于唯一标识（通常是 ID 或时间戳）进行分页，避免了 OFFSET 的性能问题。
python
from fastapi import FastAPI, Query
app: FastAPI = FastAPI()
@app.get("/items-cursor")
def get_items_cursor(
after: int | None = Query(default=None, ge=0, description="上一页最后一条记录的 ID"),
limit: int = Query(default=20, ge=1, le=100, description="每页数量"),
) -> dict[str, int | list[str] | None]:
all_items: list[str] = [f"item_{i}" for i in range(1000)]
start_idx: int = 0
if after is not None:
start_idx = after + 1 # 从游标之后开始
page_items: list[str] = all_items[start_idx : start_idx + limit]
next_cursor: int | None = None
if len(page_items) == limit and start_idx + limit < len(all_items):
next_cursor = start_idx + limit - 1
return {
"items": page_items,
"next_cursor": next_cursor,
"limit": limit,
}游标分页的工作原理：
请求: GET /items-cursor?after=29&limit=10
SQL 执行流程:
┌───────────────────────────────────────────┐
│ 1. WHERE id > 29（利用索引快速定位） │
│ 2. ORDER BY id ASC │
│ 3. LIMIT 10 │
│ │
│ [1]...[29] │ [30][31]...[39] │ [40]... │
│ 已访问 ↑ 直接定位并返回 ↑ 不关心 │
│ (索引查找) │
└───────────────────────────────────────────┘1.4 两种分页方式对比
┌──────────────────┬─────────────────────────┬─────────────────────────┐
│ 特性 │ 偏移分页 (Offset) │ 游标分页 (Cursor) │
├──────────────────┼─────────────────────────┼─────────────────────────┤
│ 适用场景 │ 需要跳页、显示总页数 │ 无限滚动、大数据集 │
│ 深层分页性能 │ 差（OFFSET 越大越慢） │ 优（始终索引查找） │
│ 数据一致性 │ 差（插入/删除导致偏移变化）│ 优（基于唯一键定位） │
│ 总页数计算 │ 支持（需要 COUNT） │ 不支持（无法预知总数） │
│ 实现复杂度 │ 简单 │ 中等 │
│ SQL 复杂度 │ LIMIT n OFFSET m │ WHERE id > ? LIMIT n │
└──────────────────┴─────────────────────────┴─────────────────────────┘第二部分：过滤与排序（L1）
2.1 查询参数过滤
实际场景：任务列表需要按状态、优先级筛选，只返回符合条件的记录。
问题：如何构建灵活的过滤查询？
python
from fastapi import FastAPI, Query
app: FastAPI = FastAPI()
# 模拟数据库
tasks: list[dict] = [
{"id": 1, "title": "任务 A", "status": "pending", "priority": "high", "assignee": "张三"},
{"id": 2, "title": "任务 B", "status": "completed", "priority": "low", "assignee": "李四"},
{"id": 3, "title": "任务 C", "status": "pending", "priority": "medium", "assignee": "张三"},
{"id": 4, "title": "任务 D", "status": "in_progress", "priority": "high", "assignee": "王五"},
]
@app.get("/tasks")
def get_tasks(
status: str | None = Query(default=None, description="状态过滤"),
priority: str | None = Query(default=None, description="优先级过滤"),
assignee: str | None = Query(default=None, description="负责人过滤"),
) -> list[dict]:
result: list[dict] = tasks
if status:
result = [t for t in result if t["status"] == status]
if priority:
result = [t for t in result if t["priority"] == priority]
if assignee:
result = [t for t in result if t["assignee"] == assignee]
return result组合查询效果：
GET /tasks?status=pending&priority=high
→ 返回: [{"id": 1, "title": "任务 A", ...}]
GET /tasks?assignee=张三
→ 返回: [任务 A, 任务 C]
GET /tasks?status=pending&priority=high&assignee=张三
→ 返回: [任务 A]2.2 搜索（全文/模糊匹配）
python
@app.get("/tasks/search")
def search_tasks(
q: str = Query(default="", min_length=0, max_length=100, description="搜索关键词"),
) -> list[dict]:
if not q.strip():
return tasks
keyword: str = q.lower()
return [
t for t in tasks
if keyword in t["title"].lower()
or keyword in t["status"].lower()
or keyword in t["assignee"].lower()
]2.3 排序
python
from typing import Literal
@app.get("/tasks/sorted")
def get_sorted_tasks(
sort: Literal["id", "title", "priority", "status"] = Query(
default="id", description="排序字段"
),
order: Literal["asc", "desc"] = Query(
default="asc", description="排序方向"
),
) -> list[dict]:
reverse: bool = order == "desc"
return sorted(tasks, key=lambda t: t[sort], reverse=reverse)2.4 组合查询：过滤 + 搜索 + 排序
请求参数组合:
┌────────────────────────────────────────────────────────────┐
│ GET /tasks │
│ ?status=pending ← 过滤 │
│ &priority=high ← 过滤 │
│ &q=紧急 ← 搜索 │
│ &sort=id ← 排序字段 │
│ &order=desc ← 排序方向 │
│ &page=1 ← 页码 │
│ &size=10 ← 每页数量 │
│ │
│ 执行顺序: 过滤 → 搜索 → 排序 → 分页 │
└────────────────────────────────────────────────────────────┘第三部分：L2 实践层
3.1 分页最佳实践
┌──────────────────────────┬────────────────────────────────────────────┐
│ 实践项 │ 推荐做法 │
├──────────────────────────┼────────────────────────────────────────────┤
│ 默认页面大小 │ 20 条（平衡性能与用户体验） │
│ 最大页面大小 │ 100 条（防止滥用） │
│ 页码起始值 │ 1（人类友好），非 0 │
│ 总数返回 │ 始终返回 total（前端需要计算总页数） │
│ 最大偏移限制 │ 限制 OFFSET <= 10000（防止深层分页） │
│ 空页处理 │ 返回空列表 []，而非 404 │
│ 页码越界 │ 返回空列表或 400 错误 │
│ 响应头 │ 可添加 X-Total-Count 等自定义头 │
└──────────────────────────┴────────────────────────────────────────────┘3.2 标准化分页响应格式
python
from pydantic import BaseModel
from typing import Generic, TypeVar
T = TypeVar("T")
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
size: int
pages: int # 总页数
@property
def has_next(self) -> bool:
return self.page < self.pages
@property
def has_prev(self) -> bool:
return self.page > 1响应示例：
json
{
"items": [
{"id": 21, "title": "任务 U", "status": "pending"},
{"id": 22, "title": "任务 V", "status": "completed"}
],
"total": 1000,
"page": 2,
"size": 20,
"pages": 50,
"has_next": true,
"has_prev": true
}3.3 反模式
┌──────────────────────────┬────────────────────────────────────────────┐
│ 反模式 │ 问题 │
├──────────────────────────┼────────────────────────────────────────────┤
│ 无限制的查询 │ SELECT * FROM items → 内存溢出、超时 │
│ 无最大页面大小限制 │ ?size=999999 → 拖垮服务 │
│ OFFSET 深层分页 │ OFFSET 1000000 → 扫描百万行只为取 10 条 │
│ 每次请求都 COUNT(*) │ 大表 COUNT 很慢，可考虑缓存或估算 │
│ 忽略排序的稳定性 │ 无 ORDER BY 的分页，结果不可预测 │
│ 客户端传入未验证的页码 │ page=-1 或 page=999999 需要边界检查 │
└──────────────────────────┴────────────────────────────────────────────┘3.4 OFFSET 深层分页性能分析
OFFSET 性能曲线（1000 万行表）:
响应时间 (ms)
│
│ ╭────── 性能拐点
│ ╭─╯
│ ╭─╯
│ ╭─╯
│ ╭─╯
│ ╭─╯
│ ╭─╯
│ ╭──╯
│ ╭─╯
│ ╭╯
│╭╯
└──────────────────────────── OFFSET 值
0 10k 100k 500k 1M 5M
原因:
1. 数据库必须扫描并丢弃 OFFSET 之前的所有行
2. OFFSET 1000000 LIMIT 10 → 扫描 1000010 行，返回 10 行
3. 99.999% 的 I/O 和 CPU 都浪费了第四部分：实战代码（L2）
4.1 完整的分页 + 过滤 + 排序端点
python
from fastapi import FastAPI, Query, HTTPException
from pydantic import BaseModel
from typing import Generic, TypeVar, Literal
app: FastAPI = FastAPI()
T = TypeVar("T")
# ==================== 分页响应模型 ====================
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
size: int
pages: int
# ==================== 数据模型 ====================
class Task(BaseModel):
id: int
title: str
status: str
priority: str
assignee: str
created_at: str
# 模拟数据库
TASKS_DB: list[Task] = [
Task(id=i, title=f"任务 {i}", status=["pending", "completed", "in_progress"][i % 3],
priority=["low", "medium", "high"][i % 3], assignee=["张三", "李四", "王五"][i % 3],
created_at=f"2024-01-{i % 28 + 1:02d}")
for i in range(1, 201)
]
# ==================== 查询构建辅助 ====================
def build_query(
status: str | None = None,
priority: str | None = None,
assignee: str | None = None,
q: str | None = None,
) -> list[Task]:
"""构建过滤后的查询结果"""
result: list[Task] = TASKS_DB
if status:
result = [t for t in result if t.status == status]
if priority:
result = [t for t in result if t.priority == priority]
if assignee:
result = [t for t in result if t.assignee == assignee]
if q:
keyword: str = q.lower()
result = [
t for t in result
if keyword in t.title.lower()
or keyword in t.assignee.lower()
]
return result
def sort_items(
items: list[Task],
sort_by: str = "id",
order: Literal["asc", "desc"] = "asc",
) -> list[Task]:
"""排序"""
reverse: bool = order == "desc"
return sorted(items, key=lambda t: getattr(t, sort_by, t.id), reverse=reverse)
def paginate(
items: list[Task],
page: int,
size: int,
) -> PaginatedResponse[Task]:
"""分页"""
total: int = len(items)
pages: int = (total + size - 1) // size if size > 0 else 0
start: int = (page - 1) * size
end: int = start + size
return PaginatedResponse(
items=items[start:end],
total=total,
page=page,
size=size,
pages=pages,
)
# ==================== 路由 ====================
@app.get("/tasks")
def get_tasks(
page: int = Query(default=1, ge=1, description="页码"),
size: int = Query(default=20, ge=1, le=100, description="每页数量"),
status: str | None = Query(default=None, description="状态过滤"),
priority: str | None = Query(default=None, description="优先级过滤"),
assignee: str | None = Query(default=None, description="负责人过滤"),
q: str | None = Query(default=None, description="搜索关键词"),
sort: Literal["id", "title", "priority", "status", "created_at"] = Query(
default="id", description="排序字段"
),
order: Literal["asc", "desc"] = Query(default="asc", description="排序方向"),
) -> PaginatedResponse[Task]:
# 1. 过滤
result: list[Task] = build_query(status, priority, assignee, q)
# 2. 排序
result = sort_items(result, sort, order)
# 3. 分页
return paginate(result, page, size)4.2 SQLAlchemy 查询构建器实现
python
from fastapi import FastAPI, Query, Depends
from sqlalchemy import select, func, Select
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import Generic, TypeVar
app: FastAPI = FastAPI()
T = TypeVar("T")
class PaginatedResponse(BaseModel, Generic[T]):
items: list[T]
total: int
page: int
size: int
pages: int
# 假设已有模型
# from models import Task, TaskResponse
def get_paginated_query(
base_query: Select,
db: Session,
page: int = 1,
size: int = 20,
) -> PaginatedResponse:
"""通用分页查询函数"""
# 获取总数
count_query: Select = select(func.count()).select_from(base_query.subquery())
total: int = db.scalar(count_query) or 0
# 分页
paginated_query: Select = base_query.offset((page - 1) * size).limit(size)
items = db.scalars(paginated_query).all()
pages: int = (total + size - 1) // size if size > 0 else 0
return PaginatedResponse(
items=items,
total=total,
page=page,
size=size,
pages=pages,
)
def build_task_query(
db: Session,
status: str | None = None,
priority: str | None = None,
q: str | None = None,
) -> Select:
"""构建任务查询"""
query: Select = select(Task)
if status:
query = query.where(Task.status == status)
if priority:
query = query.where(Task.priority == priority)
if q:
query = query.where(
Task.title.ilike(f"%{q}%") | Task.assignee.ilike(f"%{q}%")
)
return query
@app.get("/api/tasks")
def list_tasks(
page: int = Query(default=1, ge=1),
size: int = Query(default=20, ge=1, le=100),
status: str | None = Query(default=None),
priority: str | None = Query(default=None),
q: str | None = Query(default=None),
db: Session = Depends(get_db), # 假设已定义 get_db
) -> PaginatedResponse:
query: Select = build_task_query(db, status, priority, q)
query = query.order_by(Task.created_at.desc())
return get_paginated_query(query, db, page, size)4.3 游标分页完整实现
python
from fastapi import FastAPI, Query
from pydantic import BaseModel
from typing import Any
import base64
import json
app: FastAPI = FastAPI()
# ==================== 游标编码/解码 ====================
def encode_cursor(item_id: int, sort_value: str) -> str:
"""将游标信息编码为 Base64 字符串"""
data: dict[str, Any] = {"id": item_id, "sort": sort_value}
return base64.urlsafe_b64encode(json.dumps(data).encode()).decode()
def decode_cursor(cursor: str) -> dict[str, Any]:
"""解码游标字符串"""
data: bytes = base64.urlsafe_b64decode(cursor.encode())
return json.loads(data.decode())
# ==================== 游标分页响应 ====================
class CursorPageResponse(BaseModel):
items: list[dict]
next_cursor: str | None
has_more: bool
# ==================== 端点 ====================
@app.get("/items/cursor")
def get_items_cursor(
cursor: str | None = Query(default=None, description="上一页返回的游标"),
limit: int = Query(default=20, ge=1, le=100, description="每页数量"),
) -> CursorPageResponse:
all_items: list[dict] = [
{"id": i, "name": f"Item {i}", "created_at": f"2024-01-{i % 28 + 1:02d}"}
for i in range(1, 501)
]
start_idx: int = 0
if cursor:
decoded: dict[str, Any] = decode_cursor(cursor)
last_id: int = decoded["id"]
start_idx = last_id # 从游标之后开始
page_items: list[dict] = all_items[start_idx : start_idx + limit]
next_cursor: str | None = None
if len(page_items) == limit:
last_item: dict = page_items[-1]
next_cursor = encode_cursor(last_item["id"], last_item["created_at"])
return CursorPageResponse(
items=page_items,
next_cursor=next_cursor,
has_more=next_cursor is not None,
)第五部分：L3 专家层
5.1 OFFSET vs 游标分页：深度性能对比
场景: 1000 万行表，获取第 50000 页（每页 20 条）
┌─────────────────────────┬───────────────────────────┬───────────────────────────┐
│ 指标 │ OFFSET 分页 │ 游标分页 │
├─────────────────────────┼───────────────────────────┼───────────────────────────┤
│ SQL │ LIMIT 20 OFFSET 999980 │ WHERE id > 999980 │
│ │ ORDER BY id ASC LIMIT 20 │ ORDER BY id ASC │
├─────────────────────────┼───────────────────────────┼───────────────────────────┤
│ 扫描行数 │ 1,000,000 行 │ 20 行 │
│ I/O 操作 │ 全表扫描 + 排序 │ 索引查找 │
│ 响应时间（估算） │ 2-10 秒 │ < 1 毫秒 │
│ 内存占用 │ 需缓存中间结果 │ 极小 │
│ CPU 使用 │ 高（排序 + 跳过） │ 低 │
└─────────────────────────┴───────────────────────────┴───────────────────────────┘5.2 为什么 OFFSET 1000000 会慢
OFFSET 深层分页的底层机制:
数据页扫描过程
──────────────
┌───┬───┬───┬───┬───┬───┬───┬───┬───┬───┐
│ 1 │ 2 │ 3 │ 4 │ 5 │ 6 │ 7 │ 8 │ 9 │10 │ ← 页面编号
└───┴───┴───┴───┴───┴───┴───┴───┴───┴───┘
↑ ↑
开始扫描 到达 OFFSET 位置
(从这里开始 (已经扫描了
逐页读取) 999,999 行!)
┌───┐
│11 │ ← LIMIT 只取这里
└───┘
问题根源:
1. InnoDB/PostgreSQL 没有"跳到第 N 行"的快捷方式
2. 必须逐行扫描直到 OFFSET 位置
3. 排序操作（ORDER BY）会加剧性能问题
4. 没有覆盖索引时，还需要回表查询完整行数据
优化方向:
- 使用游标分页: WHERE id > last_seen_id
- 延迟关联: 先用覆盖索引找到 ID，再 JOIN 回表
- 限制最大偏移: 只允许翻到前 N 页5.3 游标分页的原理与限制
游标分页核心公式:
WHERE (sort_column, id) > (last_sort_value, last_id)
ORDER BY sort_column ASC, id ASC
LIMIT n
优势:
┌───────────────────────────────────────────────────┐
│ 1. 始终利用索引，时间复杂度 O(log N + K) │
│ 2. 不受数据量增长影响 │
│ 3. 数据插入/删除不影响分页一致性 │
│ 4. 适合无限滚动、Feed 流、日志查询 │
└───────────────────────────────────────────────────┘
限制:
┌───────────────────────────────────────────────────┐
│ 1. 无法跳转到任意页（只能上一页/下一页） │
│ 2. 无法获取总页数 │
│ 3. 排序字段必须有索引且值唯一（或联合唯一） │
│ 4. 不适合需要页码导航的传统分页 UI │
└───────────────────────────────────────────────────┘5.4 延迟关联优化 OFFSET
当必须使用 OFFSET 且无法切换到游标分页时的优化方案：
python
# 延迟关联（Deferred Join）
# 原始慢查询:
# SELECT * FROM tasks WHERE status = 'pending'
# ORDER BY created_at DESC LIMIT 20 OFFSET 1000000;
# 优化后:
# SELECT t.* FROM tasks t
# INNER JOIN (
# SELECT id FROM tasks
# WHERE status = 'pending'
# ORDER BY created_at DESC
# LIMIT 20 OFFSET 1000000
# ) AS tmp ON t.id = tmp.id;
# 原理:
# 子查询只用覆盖索引（id + created_at），不需要回表
# 确定 20 个 ID 后，再通过主键索引回表获取完整数据
# 大幅减少回表次数: 1000020 次 → 20 次5.5 知识关联图
分页与过滤知识体系
═══════════════════
┌─────────────┐
│ 分页与过滤 │
└──────┬──────┘
│
┌───────────────────┼───────────────────┐
▼ ▼ ▼
┌────────────┐ ┌────────────┐ ┌────────────┐
│ 分页策略 │ │ 过滤查询 │ │ 排序策略 │
└─────┬──────┘ └─────┬──────┘ └─────┬──────┘
│ │ │
┌────────┼────────┐ ┌────┼────┐ ┌────────┼────────┐
▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼
偏移 游标 键集 精确 模糊 全文 单字段 多字段 自定义
分页 分页 分页 匹配 搜索 搜索 排序 组合排序 排序
│ │ │ │ │ │ │ │ │
▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼ ▼
OFFSET WHERE WHERE = LIKE %% ORDER BY ORDER BY CASE WHEN
LIMIT id > (a,b) IN ILIKE 全文 ASC/DESC 多列 表达式
CURSOR > 范围
┌─────────────────────────────────────────────┐
│ 性能优化层 │
├─────────────────────────────────────────────┤
│ • 索引设计（覆盖索引、复合索引） │
│ • 延迟关联（Deferred Join） │
│ • COUNT 缓存 / 估算 │
│ • 查询结果缓存（Redis） │
│ • 分页边界限制（max_page, max_offset） │
└─────────────────────────────────────────────┘
┌─────────────────────────────────────────────┐
│ 最佳实践层 │
├─────────────────────────────────────────────┤
│ • 默认 page_size = 20, max = 100 │
│ • 始终返回 total（除非游标分页） │
│ • 过滤 → 排序 → 分页 的执行顺序 │
│ • 验证输入参数（page >= 1, 1 <= size <= 100）│
│ • 大数据集优先使用游标分页 │
│ • 需要页码导航使用偏移分页 │
└─────────────────────────────────────────────┘总结
| 知识点 | 说明 | 适用场景 |
|---|---|---|
| 偏移分页 | OFFSET n LIMIT m，简单直观 | 小数据量、需要页码导航 |
| 游标分页 | WHERE id > cursor LIMIT n，高性能 | 大数据量、无限滚动 |
| 过滤查询 | Query 参数组合过滤 | 条件筛选 |
| 搜索 | 关键词模糊匹配 | 全文/部分匹配 |
| 排序 | ORDER BY field ASC/DESC | 结果有序化 |
| PaginatedResponse | 标准化分页响应模型 | API 统一格式 |
| 延迟关联 | 子查询先取 ID 再 JOIN | 优化深层 OFFSET |
| 最大页面限制 | le=100 防止滥用 | 安全防护 |