1. 从“全量拉取”到“按需分页”的思维转变如果你正在用 FastAPI 写后端接口并且数据量稍微大一点比如超过几百条那你肯定遇到过这个问题前端一个请求过来你直接从数据库SELECT * FROM table然后一股脑儿return回去。结果就是接口响应慢得像蜗牛前端页面卡顿浏览器内存飙升用户体验直接跌到谷底。这其实就是典型的“全量拉取”思维在数据驱动的现代应用中这几乎是不可接受的。分页就是解决这个问题的标准答案它不仅仅是技术实现更是一种服务端资源管理和用户体验优化的核心设计思想。FastAPI 作为一个现代、高性能的 Python Web 框架它本身并没有内置一个叫Paginator的类来帮你搞定一切。但这恰恰是它的优势所在——它提供了足够的灵活性和高性能的异步支持让你可以根据自己的业务场景和数据库选型自由地实现最合适的分页策略。无论是传统的LIMIT/OFFSET还是基于游标的“下一页”分页或是更复杂的分页总数计算优化FastAPI 都能优雅地支撑。在接下来的内容里我不会只给你几行干巴巴的代码。我会带你深入理解分页的几种常见模式剖析它们的优缺点和适用场景然后手把手在 FastAPI 中实现它们。更重要的是我会分享在实际高并发、大数据量场景下踩过的坑和优化技巧比如为什么OFFSET在大页码时是性能杀手如何高效地获取总记录数以及如何设计一个让前后端都舒服的分页响应体。我们的目标不仅仅是“实现功能”而是“设计一个健壮、高效、可维护的分页方案”。2. 分页的核心模式LIMIT/OFFSETvs. 游标分页在动手写代码之前我们必须搞清楚两种主流的分页模式这决定了你接口的性能天花板和用户体验。2.1LIMIT/OFFSET简单直接但有明显瓶颈这是最常见、最直观的分页方式几乎所有的 SQL 数据库都支持。-- 获取第3页的数据假设每页10条 SELECT * FROM items ORDER BY id ASC LIMIT 10 OFFSET 20;工作原理LIMIT指定返回的记录数页大小OFFSET指定跳过多少条记录(页码-1) * 页大小。数据库需要先排序然后扫描并跳过OFFSET指定的行数最后返回LIMIT指定的行数。优点实现简单语法直观易于理解和实现。随机跳页用户可以随意跳转到第5页、第100页非常适合带有页码选择器的传统表格。缺点与性能陷阱OFFSET的性能问题这是最致命的缺点。OFFSET 10000意味着数据库需要先扫描并丢弃前10000条记录。随着OFFSET值的增大查询效率会线性下降尤其是在百万级数据表中跳转到靠后的页码会非常慢。数据不一致性漂移如果在分页过程中有新的数据插入到当前页之前或者当前页的数据被删除那么使用固定的OFFSET会导致某些记录被重复看到或直接跳过。例如你刚看完第1页此时新增了一条数据排在所有记录之前你再请求第2页时使用的OFFSET是10实际上跳过了11条记录导致原来第1页的最后一条记录被“挤”到了第2页而新记录则出现在了第1页。2.2 游标分页Cursor-based Pagination为“无限滚动”而生游标分页有时也叫“键集分页”Keyset Pagination是解决OFFSET性能和数据漂移问题的利器。它不依赖页码而是依赖一个唯一的、有序的“游标”通常是时间戳或自增ID。工作原理客户端不是传递页码而是传递上一页最后一条记录的“游标”值。服务端查询所有“游标”值大于或小于该值的记录并取前N条。-- 假设上一页最后一条记录的id是20获取下一页 SELECT * FROM items WHERE id 20 ORDER BY id ASC LIMIT 10;优点性能稳定WHERE id cursor这种查询可以利用索引进行高效的范围扫描无论“翻”到多后面性能都几乎恒定。数据一致性由于查询是基于某个确定的时间点或ID值在数据变动不频繁的列上能有效避免分页过程中的数据漂移问题。适合无限滚动这是移动端App如微博、Twitter信息流最常用的分页方式。缺点与限制无法随机跳页用户只能一页一页地“下一页”或“上一页”无法直接跳到第50页。对排序字段要求高游标字段必须是唯一且有序的。如果按非唯一字段如created_at时间可能重复排序需要结合另一个唯一字段如id组成复合游标实现会变复杂。实现稍复杂需要前后端约定游标的传递和解析方式。如何选择后台管理系统、数据报表等需要随机跳页、精确导航的场景用LIMIT/OFFSET。社交媒体动态、消息流、商品瀑布流等连续浏览、体验优先的场景用游标分页。3. 在 FastAPI 中实现LIMIT/OFFSET分页我们以一个简单的“文章列表”接口为例使用 SQLAlchemyORM和 SQLite/PostgreSQL 数据库。3.1 定义数据模型与依赖首先定义我们的Item模型和数据库会话。# models.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from datetime import datetime import pytz Base declarative_base() class Item(Base): __tablename__ items id Column(Integer, primary_keyTrue, indexTrue) title Column(String, indexTrue) description Column(String) created_at Column(DateTime, defaultlambda: datetime.now(pytz.UTC)) # 使用UTC时间# database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL sqlite:///./test.db # 或你的PostgreSQL连接字符串 engine create_engine(SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} if SQLALCHEMY_DATABASE_URL.startswith(sqlite) else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()3.2 设计分页请求与响应模型这是良好API设计的关键。我们使用 Pydantic 模型来定义输入和输出的“合同”。# schemas.py from pydantic import BaseModel, Field from typing import Generic, TypeVar, List, Optional from pydantic.generics import GenericModel T TypeVar(T) class PaginationParams(BaseModel): 分页查询参数 page: int Field(1, ge1, description页码从1开始) size: int Field(10, ge1, le100, description每页数量最大100) class PaginatedResponse(GenericModel, Generic[T]): 标准分页响应体 items: List[T] # 当前页的数据列表 total: int # 总记录数 page: int # 当前页码 size: int # 每页大小 pages: int # 总页数 classmethod def create(cls, items: List[T], total: int, params: PaginationParams): 快速创建分页响应的辅助方法 pages (total params.size - 1) // params.size # 向上取整计算总页数 return cls( itemsitems, totaltotal, pageparams.page, sizeparams.size, pagespages ) class ItemResponse(BaseModel): 单个项目的响应模型 id: int title: str description: Optional[str] None created_at: datetime class Config: orm_mode True # 允许从ORM模型实例直接转换注意size字段使用了le100进行限制。这是一个非常重要的安全性和性能实践。永远不要让客户端可以无限制地请求大量数据比如size10000这会导致数据库和网络瞬间过载。通常根据业务需求设置一个合理的上限如50 100 200。3.3 实现分页查询接口现在在 FastAPI 路由中实现核心的分页逻辑。# main.py from fastapi import FastAPI, Depends, Query from sqlalchemy.orm import Session from sqlalchemy import func from typing import Optional from . import models, schemas, database app FastAPI() app.get(/items/, response_modelschemas.PaginatedResponse[schemas.ItemResponse]) async def read_items( db: Session Depends(database.get_db), page_params: schemas.PaginationParams Depends(), # 依赖注入会自动从查询参数解析 title: Optional[str] Query(None, description按标题过滤) # 可选的过滤参数 ): 获取项目列表带分页和过滤 # 1. 构建基础查询 query db.query(models.Item) if title: query query.filter(models.Item.title.contains(title)) # 简单模糊查询 # 2. 计算总记录数这是一个潜在的性能点 total query.count() # 3. 应用分页计算偏移量执行查询 offset (page_params.page - 1) * page_params.size items query.order_by(models.Item.created_at.desc()).offset(offset).limit(page_params.size).all() # 4. 构建并返回标准分页响应 return schemas.PaginatedResponse.create( itemsitems, totaltotal, paramspage_params )接口调用示例GET /items/?page2size15titlefastapi响应体示例{ items: [ {id: 16, title: FastAPI 入门, ...}, {id: 17, title: FastAPI 分页实践, ...} // ... 共15条 ], total: 150, page: 2, size: 15, pages: 10 }3.4LIMIT/OFFSET的深度优化与避坑指南上面的实现是基础版但在生产环境中我们需要考虑更多。坑点一COUNT(*)的性能问题total query.count()在数据量巨大千万级且查询条件复杂时可能会非常慢因为它需要扫描所有符合条件的行。对于不需要精确总页数的场景比如无限滚动只关心“是否有下一页”可以考虑不计算总数或者使用估算值如 PostgreSQL 的pg_class.reltuples。如果必须精确计算确保WHERE条件中的字段都有索引。优化方案使用窗口函数仅限 PostgreSQL 等支持窗口函数的数据库可以一次查询同时获取分页数据和总数减少一次数据库往返。from sqlalchemy import over, func # 在查询中增加一个窗口函数来计算行号 query db.query( models.Item, func.count(models.Item.id).over().label(total) # 窗口函数计算总数 ) if title: query query.filter(models.Item.title.contains(title)) # 应用分页 items_with_total query.order_by(models.Item.created_at.desc()).offset(offset).limit(page_size).all() if items_with_total: total items_with_total[0].total # 从第一条记录中取出总数 items [item for item, _ in items_with_total] # 解包出Item对象 else: total 0 items []这个技巧在特定场景下能提升性能但窗口函数本身也有计算开销需要根据实际情况测试。坑点二OFFSET深度分页的性能悬崖如前所述OFFSET 100000是灾难。对于必须支持深度跳页且数据量大的场景一个优化思路是使用“索引覆盖扫描延迟关联”。简单说先通过一个子查询快速定位到目标页的起始主键ID只扫描索引不取数据然后再根据这些ID去取完整数据。-- 假设id是主键且有序 SELECT * FROM items WHERE id (SELECT id FROM items ORDER BY id LIMIT 1 OFFSET 100000) LIMIT 20;在 SQLAlchemy 中实现这个需要一些子查询技巧。但更根本的解决方法是与产品经理沟通是否真的需要支持用户直接跳到第5000页很多时候提供一个“输入框跳转”的功能但其背后限制一个最大可跳转页码比如100页是更合理的折中方案。4. 在 FastAPI 中实现游标分页游标分页的请求和响应模型与LIMIT/OFFSET不同。4.1 定义游标分页模型# schemas.py (追加) class CursorPaginationParams(BaseModel): 游标分页查询参数 cursor: Optional[int] Field(None, description游标通常为上一条记录的ID或时间戳) size: int Field(10, ge1, le100, description每页数量) direction: str Field(next, regex^(next|prev)$, description方向next-下一页prev-上一页) class CursorPaginatedResponse(GenericModel, Generic[T]): 游标分页响应体 items: List[T] next_cursor: Optional[int] Field(None, description用于获取下一页的游标) prev_cursor: Optional[int] Field(None, description用于获取上一页的游标) has_next: bool Field(..., description是否有下一页) has_prev: bool Field(..., description是否有上一页)4.2 实现游标分页接口这里我们以id作为游标字段实现“下一页”和“上一页”功能。# main.py (追加) app.get(/items/cursor/, response_modelschemas.CursorPaginatedResponse[schemas.ItemResponse]) async def read_items_cursor( db: Session Depends(database.get_db), params: schemas.CursorPaginationParams Depends(), title: Optional[str] Query(None) ): query db.query(models.Item) if title: query query.filter(models.Item.title.contains(title)) # 根据方向和游标构建查询条件 if params.direction next: # 获取下一页id cursor if params.cursor: query query.filter(models.Item.id params.cursor) items query.order_by(models.Item.id.asc()).limit(params.size 1).all() # 多取一条用于判断has_next else: # prev # 获取上一页id cursor并且需要逆序 if params.cursor: query query.filter(models.Item.id params.cursor) items query.order_by(models.Item.id.desc()).limit(params.size 1).all() # 多取一条用于判断has_prev items list(reversed(items)) # 将结果反转保持时间正序 # 判断是否有更多数据 has_more len(items) params.size if has_more: items items[:params.size] # 截取实际需要的数据量 # 计算前后游标 next_cursor items[-1].id if (items and params.direction next and has_more) else None prev_cursor items[0].id if (items and params.direction prev and has_more) else None # 判断是否有上一页/下一页简化逻辑实际可能更复杂 has_next next_cursor is not None has_prev prev_cursor is not None return schemas.CursorPaginatedResponse( itemsitems, next_cursornext_cursor, prev_cursorprev_cursor, has_nexthas_next, has_prevhas_prev )核心逻辑解析多取一条查询时limit(params.size 1)。这是判断是否还有下一页/上一页的关键。如果返回的记录数大于请求的size说明还有数据。方向处理next方向是id cursor正序prev方向是id cursor逆序取到结果后再反转以保证返回给客户端的列表顺序始终是一致的通常是时间倒序。游标计算next_cursor是当前页最后一条记录的IDprev_cursor是当前页第一条记录的ID。接口调用示例首次请求无游标GET /items/cursor/?size10directionnext获取下一页GET /items/cursor/?cursor25size10directionnext获取上一页GET /items/cursor/?cursor15size10directionprev4.3 游标分页的进阶挑战与解决方案挑战一基于非唯一字段如时间排序如果按created_at分页而同一时间戳可能有多条记录直接用WHERE created_at cursor会导致数据丢失或重复。解决方案是使用复合游标(created_at, id)。# 请求参数需要两个字段 cursor_time: Optional[datetime] None cursor_id: Optional[int] None # 查询条件变为 if cursor_time and cursor_id: query query.filter( (models.Item.created_at cursor_time) | ((models.Item.created_at cursor_time) (models.Item.id cursor_id)) )响应时也需要返回复合游标。这增加了前后端协议的复杂性。挑战二双向遍历与状态保持游标分页通常只适合单向连续遍历。如果用户想从第10页回到第5页客户端需要保存之前各页的游标历史或者服务端提供更复杂的机制。在实际中很多“上一页”功能在游标分页里是模拟出来的且只能回到刚刚看过的上一页而非任意历史页。挑战三过滤条件变化如果分页过程中用户改变了过滤条件如搜索关键词那么之前的游标就失效了。此时通常需要重置分页从第一页重新开始。这是游标分页与LIMIT/OFFSET相比的一个不灵活之处。5. 分页的“最后一公里”前端协作与最佳实践分页不仅仅是后端的事一个良好的分页体验需要前后端密切配合。5.1 响应头与超媒体链接HATEOAS对于 RESTful API一种更优雅的做法是在响应头或响应体中包含分页链接让客户端可以像浏览网页一样发现下一页、上一页。from fastapi.responses import JSONResponse from urllib.parse import urlencode app.get(/items/hateoas/) async def read_items_hateoas( db: Session Depends(database.get_db), page: int Query(1, ge1), size: int Query(10, ge1, le100), ): # ... 分页查询逻辑 ... total ... items ... pages ... # 构建链接 base_url fhttp://your-api.com/items/hateoas/ links { self: f{base_url}?{urlencode({page: page, size: size})}, first: f{base_url}?{urlencode({page: 1, size: size})}, last: f{base_url}?{urlencode({page: pages, size: size})} if pages 0 else None, } if page 1: links[prev] f{base_url}?{urlencode({page: page-1, size: size})} if page pages: links[next] f{base_url}?{urlencode({page: page1, size: size})} return JSONResponse( content{ items: items, pagination: { total: total, pages: pages, page: page, size: size, }, _links: links } )这遵循了 HATEOAS 原则让 API 更自描述。虽然在前端 SPA 中不一定直接使用但对 API 的消费者尤其是第三方非常友好。5.2 与前端框架的配合Element UI / Ant Design 表格它们通常需要{ total, list, page, size }这样的响应结构我们的PaginatedResponse模型可以直接对应。无限滚动组件需要{ items, next_cursor, has_next }这样的结构我们的CursorPaginatedResponse模型正好匹配。前端只需在滚动到底部时将next_cursor作为参数请求下一批数据即可。状态管理前端在 Vuex/Pinia 或 Redux 中管理分页状态时除了存储items还应存储当前的page/size或cursor以及total等信息以便在路由变化或组件销毁重建时能恢复分页状态。5.3 性能监控与调试慢查询日志务必在数据库和 ORM 层面开启慢查询日志监控那些OFFSET值巨大或COUNT很慢的查询。API 响应时间监控关注分页接口的 P95、P99 响应时间特别是随着页码增大的性能衰减曲线。使用EXPLAIN ANALYZE对于复杂的分析型分页查询定期使用EXPLAIN命令分析执行计划确保索引被正确使用。6. 总结与个人实战心得分页功能初看简单但想在生产环境中做得稳健、高效需要考虑的细节非常多。回顾一下核心要点模式选择是第一要务在项目初期就和产品、前端确定好交互模式。需要随机跳页选LIMIT/OFFSET需要连续流畅浏览选游标分页。不要试图用一个接口满足所有场景。OFFSET是性能毒药对于大数据集尽量避免深度跳页。如果业务必须考虑使用“索引覆盖查询”优化或者用业务逻辑限制最大可访问页码。COUNT(*)可能很重评估是否真的需要精确的总数。对于无限滚动has_next布尔值就够了。如果需要确保过滤条件有索引或探索数据库的估算功能。游标分页的游标要稳定优先使用自增主键或具有唯一性的时间戳。按非唯一字段分页会引入复杂性。API 设计要规范使用清晰的请求/响应模型Pydantic对参数进行验证如size的最大值限制。考虑加入 HATEOAS 链接提升 API 可发现性。索引是性能的基石确保ORDER BY、WHERE以及作为游标的字段上建立了合适的索引。对于复合排序和过滤可能需要复合索引。在我经历的一个项目中初期使用了简单的LIMIT/OFFSET当用户表增长到百万级后管理员在后台查看最后一页的用户列表时接口超时。我们将这个特定场景改造成了游标分页因为管理员通常也是逐页审核并保留了其他需要跳页的报表功能使用LIMIT/OFFSET但增加了页码上限。同时我们为常用的查询组合创建了复合索引并将一些不必要精确计算总数的列表页的COUNT查询移除改为只判断has_next。这些组合拳下来相关接口的 P99 延迟下降了 90% 以上。最后记住没有银弹。最好的分页策略是贴合你的具体业务需求、数据规模和用户行为。在 FastAPI 这个灵活的框架下你有足够的工具去实现和优化它。希望这篇长文能帮你避开我踩过的那些坑构建出既快又稳的分页功能。
FastAPI分页实战：从LIMIT/OFFSET到游标分页的深度解析
1. 从“全量拉取”到“按需分页”的思维转变如果你正在用 FastAPI 写后端接口并且数据量稍微大一点比如超过几百条那你肯定遇到过这个问题前端一个请求过来你直接从数据库SELECT * FROM table然后一股脑儿return回去。结果就是接口响应慢得像蜗牛前端页面卡顿浏览器内存飙升用户体验直接跌到谷底。这其实就是典型的“全量拉取”思维在数据驱动的现代应用中这几乎是不可接受的。分页就是解决这个问题的标准答案它不仅仅是技术实现更是一种服务端资源管理和用户体验优化的核心设计思想。FastAPI 作为一个现代、高性能的 Python Web 框架它本身并没有内置一个叫Paginator的类来帮你搞定一切。但这恰恰是它的优势所在——它提供了足够的灵活性和高性能的异步支持让你可以根据自己的业务场景和数据库选型自由地实现最合适的分页策略。无论是传统的LIMIT/OFFSET还是基于游标的“下一页”分页或是更复杂的分页总数计算优化FastAPI 都能优雅地支撑。在接下来的内容里我不会只给你几行干巴巴的代码。我会带你深入理解分页的几种常见模式剖析它们的优缺点和适用场景然后手把手在 FastAPI 中实现它们。更重要的是我会分享在实际高并发、大数据量场景下踩过的坑和优化技巧比如为什么OFFSET在大页码时是性能杀手如何高效地获取总记录数以及如何设计一个让前后端都舒服的分页响应体。我们的目标不仅仅是“实现功能”而是“设计一个健壮、高效、可维护的分页方案”。2. 分页的核心模式LIMIT/OFFSETvs. 游标分页在动手写代码之前我们必须搞清楚两种主流的分页模式这决定了你接口的性能天花板和用户体验。2.1LIMIT/OFFSET简单直接但有明显瓶颈这是最常见、最直观的分页方式几乎所有的 SQL 数据库都支持。-- 获取第3页的数据假设每页10条 SELECT * FROM items ORDER BY id ASC LIMIT 10 OFFSET 20;工作原理LIMIT指定返回的记录数页大小OFFSET指定跳过多少条记录(页码-1) * 页大小。数据库需要先排序然后扫描并跳过OFFSET指定的行数最后返回LIMIT指定的行数。优点实现简单语法直观易于理解和实现。随机跳页用户可以随意跳转到第5页、第100页非常适合带有页码选择器的传统表格。缺点与性能陷阱OFFSET的性能问题这是最致命的缺点。OFFSET 10000意味着数据库需要先扫描并丢弃前10000条记录。随着OFFSET值的增大查询效率会线性下降尤其是在百万级数据表中跳转到靠后的页码会非常慢。数据不一致性漂移如果在分页过程中有新的数据插入到当前页之前或者当前页的数据被删除那么使用固定的OFFSET会导致某些记录被重复看到或直接跳过。例如你刚看完第1页此时新增了一条数据排在所有记录之前你再请求第2页时使用的OFFSET是10实际上跳过了11条记录导致原来第1页的最后一条记录被“挤”到了第2页而新记录则出现在了第1页。2.2 游标分页Cursor-based Pagination为“无限滚动”而生游标分页有时也叫“键集分页”Keyset Pagination是解决OFFSET性能和数据漂移问题的利器。它不依赖页码而是依赖一个唯一的、有序的“游标”通常是时间戳或自增ID。工作原理客户端不是传递页码而是传递上一页最后一条记录的“游标”值。服务端查询所有“游标”值大于或小于该值的记录并取前N条。-- 假设上一页最后一条记录的id是20获取下一页 SELECT * FROM items WHERE id 20 ORDER BY id ASC LIMIT 10;优点性能稳定WHERE id cursor这种查询可以利用索引进行高效的范围扫描无论“翻”到多后面性能都几乎恒定。数据一致性由于查询是基于某个确定的时间点或ID值在数据变动不频繁的列上能有效避免分页过程中的数据漂移问题。适合无限滚动这是移动端App如微博、Twitter信息流最常用的分页方式。缺点与限制无法随机跳页用户只能一页一页地“下一页”或“上一页”无法直接跳到第50页。对排序字段要求高游标字段必须是唯一且有序的。如果按非唯一字段如created_at时间可能重复排序需要结合另一个唯一字段如id组成复合游标实现会变复杂。实现稍复杂需要前后端约定游标的传递和解析方式。如何选择后台管理系统、数据报表等需要随机跳页、精确导航的场景用LIMIT/OFFSET。社交媒体动态、消息流、商品瀑布流等连续浏览、体验优先的场景用游标分页。3. 在 FastAPI 中实现LIMIT/OFFSET分页我们以一个简单的“文章列表”接口为例使用 SQLAlchemyORM和 SQLite/PostgreSQL 数据库。3.1 定义数据模型与依赖首先定义我们的Item模型和数据库会话。# models.py from sqlalchemy import Column, Integer, String, DateTime from sqlalchemy.ext.declarative import declarative_base from datetime import datetime import pytz Base declarative_base() class Item(Base): __tablename__ items id Column(Integer, primary_keyTrue, indexTrue) title Column(String, indexTrue) description Column(String) created_at Column(DateTime, defaultlambda: datetime.now(pytz.UTC)) # 使用UTC时间# database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL sqlite:///./test.db # 或你的PostgreSQL连接字符串 engine create_engine(SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} if SQLALCHEMY_DATABASE_URL.startswith(sqlite) else {}) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()3.2 设计分页请求与响应模型这是良好API设计的关键。我们使用 Pydantic 模型来定义输入和输出的“合同”。# schemas.py from pydantic import BaseModel, Field from typing import Generic, TypeVar, List, Optional from pydantic.generics import GenericModel T TypeVar(T) class PaginationParams(BaseModel): 分页查询参数 page: int Field(1, ge1, description页码从1开始) size: int Field(10, ge1, le100, description每页数量最大100) class PaginatedResponse(GenericModel, Generic[T]): 标准分页响应体 items: List[T] # 当前页的数据列表 total: int # 总记录数 page: int # 当前页码 size: int # 每页大小 pages: int # 总页数 classmethod def create(cls, items: List[T], total: int, params: PaginationParams): 快速创建分页响应的辅助方法 pages (total params.size - 1) // params.size # 向上取整计算总页数 return cls( itemsitems, totaltotal, pageparams.page, sizeparams.size, pagespages ) class ItemResponse(BaseModel): 单个项目的响应模型 id: int title: str description: Optional[str] None created_at: datetime class Config: orm_mode True # 允许从ORM模型实例直接转换注意size字段使用了le100进行限制。这是一个非常重要的安全性和性能实践。永远不要让客户端可以无限制地请求大量数据比如size10000这会导致数据库和网络瞬间过载。通常根据业务需求设置一个合理的上限如50 100 200。3.3 实现分页查询接口现在在 FastAPI 路由中实现核心的分页逻辑。# main.py from fastapi import FastAPI, Depends, Query from sqlalchemy.orm import Session from sqlalchemy import func from typing import Optional from . import models, schemas, database app FastAPI() app.get(/items/, response_modelschemas.PaginatedResponse[schemas.ItemResponse]) async def read_items( db: Session Depends(database.get_db), page_params: schemas.PaginationParams Depends(), # 依赖注入会自动从查询参数解析 title: Optional[str] Query(None, description按标题过滤) # 可选的过滤参数 ): 获取项目列表带分页和过滤 # 1. 构建基础查询 query db.query(models.Item) if title: query query.filter(models.Item.title.contains(title)) # 简单模糊查询 # 2. 计算总记录数这是一个潜在的性能点 total query.count() # 3. 应用分页计算偏移量执行查询 offset (page_params.page - 1) * page_params.size items query.order_by(models.Item.created_at.desc()).offset(offset).limit(page_params.size).all() # 4. 构建并返回标准分页响应 return schemas.PaginatedResponse.create( itemsitems, totaltotal, paramspage_params )接口调用示例GET /items/?page2size15titlefastapi响应体示例{ items: [ {id: 16, title: FastAPI 入门, ...}, {id: 17, title: FastAPI 分页实践, ...} // ... 共15条 ], total: 150, page: 2, size: 15, pages: 10 }3.4LIMIT/OFFSET的深度优化与避坑指南上面的实现是基础版但在生产环境中我们需要考虑更多。坑点一COUNT(*)的性能问题total query.count()在数据量巨大千万级且查询条件复杂时可能会非常慢因为它需要扫描所有符合条件的行。对于不需要精确总页数的场景比如无限滚动只关心“是否有下一页”可以考虑不计算总数或者使用估算值如 PostgreSQL 的pg_class.reltuples。如果必须精确计算确保WHERE条件中的字段都有索引。优化方案使用窗口函数仅限 PostgreSQL 等支持窗口函数的数据库可以一次查询同时获取分页数据和总数减少一次数据库往返。from sqlalchemy import over, func # 在查询中增加一个窗口函数来计算行号 query db.query( models.Item, func.count(models.Item.id).over().label(total) # 窗口函数计算总数 ) if title: query query.filter(models.Item.title.contains(title)) # 应用分页 items_with_total query.order_by(models.Item.created_at.desc()).offset(offset).limit(page_size).all() if items_with_total: total items_with_total[0].total # 从第一条记录中取出总数 items [item for item, _ in items_with_total] # 解包出Item对象 else: total 0 items []这个技巧在特定场景下能提升性能但窗口函数本身也有计算开销需要根据实际情况测试。坑点二OFFSET深度分页的性能悬崖如前所述OFFSET 100000是灾难。对于必须支持深度跳页且数据量大的场景一个优化思路是使用“索引覆盖扫描延迟关联”。简单说先通过一个子查询快速定位到目标页的起始主键ID只扫描索引不取数据然后再根据这些ID去取完整数据。-- 假设id是主键且有序 SELECT * FROM items WHERE id (SELECT id FROM items ORDER BY id LIMIT 1 OFFSET 100000) LIMIT 20;在 SQLAlchemy 中实现这个需要一些子查询技巧。但更根本的解决方法是与产品经理沟通是否真的需要支持用户直接跳到第5000页很多时候提供一个“输入框跳转”的功能但其背后限制一个最大可跳转页码比如100页是更合理的折中方案。4. 在 FastAPI 中实现游标分页游标分页的请求和响应模型与LIMIT/OFFSET不同。4.1 定义游标分页模型# schemas.py (追加) class CursorPaginationParams(BaseModel): 游标分页查询参数 cursor: Optional[int] Field(None, description游标通常为上一条记录的ID或时间戳) size: int Field(10, ge1, le100, description每页数量) direction: str Field(next, regex^(next|prev)$, description方向next-下一页prev-上一页) class CursorPaginatedResponse(GenericModel, Generic[T]): 游标分页响应体 items: List[T] next_cursor: Optional[int] Field(None, description用于获取下一页的游标) prev_cursor: Optional[int] Field(None, description用于获取上一页的游标) has_next: bool Field(..., description是否有下一页) has_prev: bool Field(..., description是否有上一页)4.2 实现游标分页接口这里我们以id作为游标字段实现“下一页”和“上一页”功能。# main.py (追加) app.get(/items/cursor/, response_modelschemas.CursorPaginatedResponse[schemas.ItemResponse]) async def read_items_cursor( db: Session Depends(database.get_db), params: schemas.CursorPaginationParams Depends(), title: Optional[str] Query(None) ): query db.query(models.Item) if title: query query.filter(models.Item.title.contains(title)) # 根据方向和游标构建查询条件 if params.direction next: # 获取下一页id cursor if params.cursor: query query.filter(models.Item.id params.cursor) items query.order_by(models.Item.id.asc()).limit(params.size 1).all() # 多取一条用于判断has_next else: # prev # 获取上一页id cursor并且需要逆序 if params.cursor: query query.filter(models.Item.id params.cursor) items query.order_by(models.Item.id.desc()).limit(params.size 1).all() # 多取一条用于判断has_prev items list(reversed(items)) # 将结果反转保持时间正序 # 判断是否有更多数据 has_more len(items) params.size if has_more: items items[:params.size] # 截取实际需要的数据量 # 计算前后游标 next_cursor items[-1].id if (items and params.direction next and has_more) else None prev_cursor items[0].id if (items and params.direction prev and has_more) else None # 判断是否有上一页/下一页简化逻辑实际可能更复杂 has_next next_cursor is not None has_prev prev_cursor is not None return schemas.CursorPaginatedResponse( itemsitems, next_cursornext_cursor, prev_cursorprev_cursor, has_nexthas_next, has_prevhas_prev )核心逻辑解析多取一条查询时limit(params.size 1)。这是判断是否还有下一页/上一页的关键。如果返回的记录数大于请求的size说明还有数据。方向处理next方向是id cursor正序prev方向是id cursor逆序取到结果后再反转以保证返回给客户端的列表顺序始终是一致的通常是时间倒序。游标计算next_cursor是当前页最后一条记录的IDprev_cursor是当前页第一条记录的ID。接口调用示例首次请求无游标GET /items/cursor/?size10directionnext获取下一页GET /items/cursor/?cursor25size10directionnext获取上一页GET /items/cursor/?cursor15size10directionprev4.3 游标分页的进阶挑战与解决方案挑战一基于非唯一字段如时间排序如果按created_at分页而同一时间戳可能有多条记录直接用WHERE created_at cursor会导致数据丢失或重复。解决方案是使用复合游标(created_at, id)。# 请求参数需要两个字段 cursor_time: Optional[datetime] None cursor_id: Optional[int] None # 查询条件变为 if cursor_time and cursor_id: query query.filter( (models.Item.created_at cursor_time) | ((models.Item.created_at cursor_time) (models.Item.id cursor_id)) )响应时也需要返回复合游标。这增加了前后端协议的复杂性。挑战二双向遍历与状态保持游标分页通常只适合单向连续遍历。如果用户想从第10页回到第5页客户端需要保存之前各页的游标历史或者服务端提供更复杂的机制。在实际中很多“上一页”功能在游标分页里是模拟出来的且只能回到刚刚看过的上一页而非任意历史页。挑战三过滤条件变化如果分页过程中用户改变了过滤条件如搜索关键词那么之前的游标就失效了。此时通常需要重置分页从第一页重新开始。这是游标分页与LIMIT/OFFSET相比的一个不灵活之处。5. 分页的“最后一公里”前端协作与最佳实践分页不仅仅是后端的事一个良好的分页体验需要前后端密切配合。5.1 响应头与超媒体链接HATEOAS对于 RESTful API一种更优雅的做法是在响应头或响应体中包含分页链接让客户端可以像浏览网页一样发现下一页、上一页。from fastapi.responses import JSONResponse from urllib.parse import urlencode app.get(/items/hateoas/) async def read_items_hateoas( db: Session Depends(database.get_db), page: int Query(1, ge1), size: int Query(10, ge1, le100), ): # ... 分页查询逻辑 ... total ... items ... pages ... # 构建链接 base_url fhttp://your-api.com/items/hateoas/ links { self: f{base_url}?{urlencode({page: page, size: size})}, first: f{base_url}?{urlencode({page: 1, size: size})}, last: f{base_url}?{urlencode({page: pages, size: size})} if pages 0 else None, } if page 1: links[prev] f{base_url}?{urlencode({page: page-1, size: size})} if page pages: links[next] f{base_url}?{urlencode({page: page1, size: size})} return JSONResponse( content{ items: items, pagination: { total: total, pages: pages, page: page, size: size, }, _links: links } )这遵循了 HATEOAS 原则让 API 更自描述。虽然在前端 SPA 中不一定直接使用但对 API 的消费者尤其是第三方非常友好。5.2 与前端框架的配合Element UI / Ant Design 表格它们通常需要{ total, list, page, size }这样的响应结构我们的PaginatedResponse模型可以直接对应。无限滚动组件需要{ items, next_cursor, has_next }这样的结构我们的CursorPaginatedResponse模型正好匹配。前端只需在滚动到底部时将next_cursor作为参数请求下一批数据即可。状态管理前端在 Vuex/Pinia 或 Redux 中管理分页状态时除了存储items还应存储当前的page/size或cursor以及total等信息以便在路由变化或组件销毁重建时能恢复分页状态。5.3 性能监控与调试慢查询日志务必在数据库和 ORM 层面开启慢查询日志监控那些OFFSET值巨大或COUNT很慢的查询。API 响应时间监控关注分页接口的 P95、P99 响应时间特别是随着页码增大的性能衰减曲线。使用EXPLAIN ANALYZE对于复杂的分析型分页查询定期使用EXPLAIN命令分析执行计划确保索引被正确使用。6. 总结与个人实战心得分页功能初看简单但想在生产环境中做得稳健、高效需要考虑的细节非常多。回顾一下核心要点模式选择是第一要务在项目初期就和产品、前端确定好交互模式。需要随机跳页选LIMIT/OFFSET需要连续流畅浏览选游标分页。不要试图用一个接口满足所有场景。OFFSET是性能毒药对于大数据集尽量避免深度跳页。如果业务必须考虑使用“索引覆盖查询”优化或者用业务逻辑限制最大可访问页码。COUNT(*)可能很重评估是否真的需要精确的总数。对于无限滚动has_next布尔值就够了。如果需要确保过滤条件有索引或探索数据库的估算功能。游标分页的游标要稳定优先使用自增主键或具有唯一性的时间戳。按非唯一字段分页会引入复杂性。API 设计要规范使用清晰的请求/响应模型Pydantic对参数进行验证如size的最大值限制。考虑加入 HATEOAS 链接提升 API 可发现性。索引是性能的基石确保ORDER BY、WHERE以及作为游标的字段上建立了合适的索引。对于复合排序和过滤可能需要复合索引。在我经历的一个项目中初期使用了简单的LIMIT/OFFSET当用户表增长到百万级后管理员在后台查看最后一页的用户列表时接口超时。我们将这个特定场景改造成了游标分页因为管理员通常也是逐页审核并保留了其他需要跳页的报表功能使用LIMIT/OFFSET但增加了页码上限。同时我们为常用的查询组合创建了复合索引并将一些不必要精确计算总数的列表页的COUNT查询移除改为只判断has_next。这些组合拳下来相关接口的 P99 延迟下降了 90% 以上。最后记住没有银弹。最好的分页策略是贴合你的具体业务需求、数据规模和用户行为。在 FastAPI 这个灵活的框架下你有足够的工具去实现和优化它。希望这篇长文能帮你避开我踩过的那些坑构建出既快又稳的分页功能。
