当前位置：首页 > news >正文

FastAPI分页实战：从LIMIT/OFFSET到游标分页的深度解析

news 2026/8/1 14:50:03

1. 从“全量拉取”到“按需分页”的思维转变

如果你正在用 FastAPI 写后端接口，并且数据量稍微大一点，比如超过几百条，那你肯定遇到过这个问题：前端一个请求过来，你直接从数据库SELECT * FROM table，然后一股脑儿return回去。结果就是，接口响应慢得像蜗牛，前端页面卡顿，浏览器内存飙升，用户体验直接跌到谷底。这其实就是典型的“全量拉取”思维，在数据驱动的现代应用中，这几乎是不可接受的。分页，就是解决这个问题的标准答案，它不仅仅是技术实现，更是一种服务端资源管理和用户体验优化的核心设计思想。

FastAPI 作为一个现代、高性能的 Python Web 框架，它本身并没有内置一个叫Paginator的类来帮你搞定一切。但这恰恰是它的优势所在——它提供了足够的灵活性和高性能的异步支持，让你可以根据自己的业务场景和数据库选型，自由地实现最合适的分页策略。无论是传统的LIMIT/OFFSET，还是基于游标的“下一页”分页，或是更复杂的分页总数计算优化，FastAPI 都能优雅地支撑。

在接下来的内容里，我不会只给你几行干巴巴的代码。我会带你深入理解分页的几种常见模式，剖析它们的优缺点和适用场景，然后手把手在 FastAPI 中实现它们。更重要的是，我会分享在实际高并发、大数据量场景下踩过的坑和优化技巧，比如为什么OFFSET在大页码时是性能杀手，如何高效地获取总记录数，以及如何设计一个让前后端都舒服的分页响应体。我们的目标不仅仅是“实现功能”，而是“设计一个健壮、高效、可维护的分页方案”。

2. 分页的核心模式：`LIMIT/OFFSET`vs. 游标分页

在动手写代码之前，我们必须搞清楚两种主流的分页模式，这决定了你接口的性能天花板和用户体验。

2.1`LIMIT/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`分页

我们以一个简单的“文章列表”接口为例，使用 SQLAlchemy（ORM）和 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_key=True, index=True) title = Column(String, index=True) description = Column(String) created_at = Column(DateTime, default=lambda: 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(autocommit=False, autoflush=False, bind=engine) 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, ge=1, description="页码，从1开始") size: int = Field(10, ge=1, le=100, 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( items=items, total=total, page=params.page, size=params.size, pages=pages ) class ItemResponse(BaseModel): """单个项目的响应模型""" id: int title: str description: Optional[str] = None created_at: datetime class Config: orm_mode = True # 允许从ORM模型实例直接转换

注意：size字段使用了le=100进行限制。这是一个非常重要的安全性和性能实践。永远不要让客户端可以无限制地请求大量数据（比如size=10000），这会导致数据库和网络瞬间过载。通常根据业务需求设置一个合理的上限（如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_model=schemas.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( items=items, total=total, params=page_params )

接口调用示例：GET /items/?page=2&size=15&title=fastapi

响应体示例：

{ "items": [ {"id": 16, "title": "FastAPI 入门", ...}, {"id": 17, "title": "FastAPI 分页实践", ...} // ... 共15条 ], "total": 150, "page": 2, "size": 15, "pages": 10 }

3.4`LIMIT/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, ge=1, le=100, 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_model=schemas.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( items=items, next_cursor=next_cursor, prev_cursor=prev_cursor, has_next=has_next, has_prev=has_prev )

核心逻辑解析：

多取一条：查询时limit(params.size + 1)。这是判断是否还有下一页/上一页的关键。如果返回的记录数大于请求的size，说明还有数据。
方向处理：next方向是id > cursor正序；prev方向是id < cursor逆序，取到结果后再反转，以保证返回给客户端的列表顺序始终是一致的（通常是时间倒序）。
游标计算：next_cursor是当前页最后一条记录的ID，prev_cursor是当前页第一条记录的ID。

接口调用示例：

首次请求（无游标）：GET /items/cursor/?size=10&direction=next
获取下一页：GET /items/cursor/?cursor=25&size=10&direction=next
获取上一页：GET /items/cursor/?cursor=15&size=10&direction=prev

4.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, ge=1), size: int = Query(10, ge=1, le=100), ): # ... 分页查询逻辑 ... total = ... items = ... pages = ... # 构建链接 base_url = f"http://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': page+1, '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 这个灵活的框架下，你有足够的工具去实现和优化它。希望这篇长文能帮你避开我踩过的那些坑，构建出既快又稳的分页功能。

查看全文

http://www.cnnetsun.cn/news/3783898.html