这次我们来看一个 FastAPI 实现前后端分页的实战项目。对于任何需要处理列表数据的 Web 应用分页都是核心功能它直接关系到用户体验和服务器性能。这个项目的重点不是概念多复杂而是如何用 FastAPI 快速、优雅地搭建一套可复用的分页接口并让前端能轻松对接。如果你关心如何在后端高效处理大数据集、如何设计标准化的分页响应格式、以及如何避免前端一次性加载过多数据导致的卡顿这篇文章可以直接收藏。我们将从零开始构建一个包含完整分页逻辑的 FastAPI 应用涵盖数据库查询、参数验证、响应模型设计以及前端调用示例。整个过程不依赖复杂的框架代码清晰易于集成到你的现有项目中。1. 核心能力速览能力项说明技术栈FastAPI (后端API框架) SQLAlchemy (ORM) Pydantic (数据验证)核心功能实现标准化的后端分页查询接口支持页码/页大小、排序、过滤等常见参数。接口设计RESTful 风格返回结构化的分页数据列表、总数、当前页、总页数等。性能考量通过数据库的LIMIT和OFFSET或更优的keyset pagination实现高效查询避免全表扫描。前端适配提供通用的 JSON 响应格式方便 Vue/React/Angular 等前端框架直接使用。启动方式通过uvicorn命令一键启动本地开发服务器。适合场景需要展示用户列表、商品列表、订单记录、日志数据等任何分页需求的 Web 应用后端开发。2. 适用场景与使用边界适合谁全栈开发者需要快速为前端提供分页数据接口。后端工程师希望构建一套标准、可维护的分页逻辑避免在每个接口重复造轮子。学习者想通过一个完整案例深入理解 FastAPI 的依赖注入、响应模型和数据库交互。能解决什么问题性能问题当数据库表数据量巨大时一次性查询所有数据会导致内存溢出、响应缓慢。分页接口只返回当前页所需的数据。体验问题前端无限滚动或页码切换需要结构化的分页信息如总条数、总页数来渲染分页器。标准化问题团队内不同接口的分页响应格式不统一增加前端对接成本。不适合什么场景实时流式数据如股票行情、聊天消息更适合使用 WebSocket 进行推送。极少量数据如果数据量固定且很少例如少于50条一次性返回可能更简单。需要复杂游标分页的场景本文主要基于页码-页大小模式对于深度分页优化如基于时间戳或ID的游标分页仅作原理性提及。使用边界与合规提醒确保分页查询涉及的数据查询符合隐私和数据安全法规避免通过分页参数恶意遍历非授权数据。在公开API中需要对分页参数如page_size设置合理的上限防止DoS攻击。3. 环境准备与前置条件在开始编码前请确保你的开发环境满足以下要求。这是一个通用清单你可以根据实际项目调整。操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python 版本Python 3.8 或更高版本。推荐使用 Python 3.10 以获得更好的类型提示支持。包管理工具pip通常随 Python 安装。核心依赖fastapi: 用于构建 API。uvicorn: 用于运行 ASGI 服务器。sqlalchemy: 用于数据库 ORM 操作。pydantic: 用于数据验证和设置管理。databases或asyncpg/aiomysql: 用于异步数据库驱动本文示例使用同步的sqlalchemy以简化但会说明异步方案。数据库SQLite用于演示、PostgreSQL 或 MySQL用于生产。可选工具pipenv或poetry: 用于虚拟环境和依赖管理。pgadmin/dbeaver: 数据库图形化管理工具。curl/Postman/Insomnia: 用于 API 测试。检查清单打开终端或 CMD/PowerShell运行python --version确认 Python 版本。运行pip --version确认 pip 可用。准备一个干净的项目目录例如fastapi-pagination-demo。4. 安装部署与启动方式我们将创建一个最小化的 FastAPI 应用来演示分页。首先在项目目录中安装依赖。4.1 创建虚拟环境与安装依赖建议使用虚拟环境隔离项目依赖。# 进入项目目录 cd fastapi-pagination-demo # 创建虚拟环境 (Windows) python -m venv venv # 激活虚拟环境 (Windows) venv\Scripts\activate # 创建虚拟环境 (macOS/Linux) python3 -m venv venv # 激活虚拟环境 (macOS/Linux) source venv/bin/activate虚拟环境激活后终端提示符前通常会出现(venv)标识。接着安装核心包pip install fastapi uvicorn sqlalchemy pydantic为了演示我们使用 SQLite 数据库它无需额外安装驱动。如果你计划使用 PostgreSQL可以安装asyncpg和databases。4.2 项目结构初始化创建以下文件和目录结构fastapi-pagination-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和路由 │ ├── database.py # 数据库连接和会话管理 │ ├── models.py # SQLAlchemy 数据模型 │ ├── schemas.py # Pydantic 响应/请求模型 │ ├── crud.py # 增删改查操作包含分页逻辑 │ └── dependencies.py # 依赖项如分页参数依赖 ├── requirements.txt └── README.md4.3 核心代码实现下面我们分步骤填充核心文件。1. 数据库配置 (app/database.py)这里我们使用 SQLAlchemy 的同步引擎进行演示。对于生产环境强烈建议使用异步引擎 (asyncpgsqlalchemy.ext.asyncio)。from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # SQLite 数据库URL数据库文件将位于项目根目录的 sql_app.db SQLALCHEMY_DATABASE_URL sqlite:///./sql_app.db # 如果是 PostgreSQLURL类似 postgresql://user:passwordlocalhost/dbname # create_engine 的参数 connect_args 仅 SQLite 需要 engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) # 创建配置过的 SessionLocal 类 SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 创建 DeclarativeMeta 基类后续模型将继承它 Base declarative_base() # 依赖项获取数据库会话 def get_db(): db SessionLocal() try: yield db finally: db.close()2. 数据模型 (app/models.py)创建一个简单的User模型作为分页的数据源。from sqlalchemy import Column, Integer, String from .database import Base class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) username Column(String, uniqueTrue, indexTrue, nullableFalse) email Column(String, uniqueTrue, indexTrue, nullableFalse) full_name Column(String)3. Pydantic 模型 (app/schemas.py)定义请求和响应的数据结构。这是实现标准化分页响应的关键。from pydantic import BaseModel from typing import List, Optional, Any # 用户的基础模型 class UserBase(BaseModel): username: str email: str full_name: Optional[str] None class UserCreate(UserBase): pass class User(UserBase): id: int class Config: orm_mode True # 允许从 ORM 对象创建 Pydantic 模型 # --- 分页相关模型 --- # 通用的分页参数模型将由依赖项使用 class PaginationParams(BaseModel): page: int 1 size: int 10 class Config: # 允许通过查询参数如 ?page1size10传入 # 在依赖项中我们会用 Depends() 来接收而不是直接作为请求体 # 这里配置主要是为了文档和可能的其他用途 pass # 通用的分页响应模型 class PaginatedResponse(BaseModel): items: List[Any] # 当前页的数据列表 total: int # 数据总条数 page: int # 当前页码 size: int # 每页大小 pages: int # 总页数 classmethod def create(cls, items: List[Any], total: int, page: int, size: int): 便捷的创建方法自动计算总页数 pages (total size - 1) // size if size 0 else 0 return cls(itemsitems, totaltotal, pagepage, sizesize, pagespages)4. 分页参数依赖 (app/dependencies.py)我们将分页参数page和size提取为依赖项这样任何需要分页的接口都可以直接注入代码更清晰。from fastapi import Query from .schemas import PaginationParams def get_pagination_params( page: int Query(1, ge1, description页码从1开始), size: int Query(10, ge1, le100, description每页数量最大100) ) - PaginationParams: 分页参数依赖项。 从查询参数中获取 page 和 size并返回 PaginationParams 对象。 ge1 确保参数 1, le100 限制 size 最大为100防止过大查询。 return PaginationParams(pagepage, sizesize)5. 数据操作与分页逻辑 (app/crud.py)这里封装了用户相关的数据库操作核心是get_users分页函数。from sqlalchemy.orm import Session from . import models, schemas from sqlalchemy import func def create_user(db: Session, user: schemas.UserCreate): db_user models.User(**user.dict()) db.add(db_user) db.commit() db.refresh(db_user) return db_user def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id user_id).first() def get_users(db: Session, skip: int 0, limit: int 100): # 简单的 limit/offset 分页适用于数据量不大或深度分页不频繁的场景 return db.query(models.User).offset(skip).limit(limit).all() def get_users_with_pagination(db: Session, pagination: schemas.PaginationParams): 带完整分页信息的查询。 返回 (当前页的用户列表, 用户总数) # 计算 offset offset (pagination.page - 1) * pagination.size # 查询当前页数据 items db.query(models.User).offset(offset).limit(pagination.size).all() # 查询总条数这是一个单独的 count 查询在数据量大时需考虑性能 total db.query(func.count(models.User.id)).scalar() return items, total6. 主应用与路由 (app/main.py)将所有部分组合起来创建 FastAPI 应用并定义路由。from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from . import crud, models, schemas, dependencies from .database import engine, get_db # 创建数据库表仅演示生产环境请使用迁移工具如 Alembic models.Base.metadata.create_all(bindengine) app FastAPI(titleFastAPI 分页演示, version1.0.0) app.post(/users/, response_modelschemas.User) def create_user(user: schemas.UserCreate, db: Session Depends(get_db)): 创建新用户。 db_user crud.get_user_by_email(db, emailuser.email) # 需要实现 get_user_by_email if db_user: raise HTTPException(status_code400, detailEmail already registered) return crud.create_user(dbdb, useruser) app.get(/users/{user_id}, response_modelschemas.User) def read_user(user_id: int, db: Session Depends(get_db)): 根据ID获取单个用户。 db_user crud.get_user(db, user_iduser_id) if db_user is None: raise HTTPException(status_code404, detailUser not found) return db_user app.get(/users/, response_modelschemas.PaginatedResponse[schemas.User]) def read_users( pagination: schemas.PaginationParams Depends(dependencies.get_pagination_params), db: Session Depends(get_db) ): 获取用户列表带分页。 这是本文的核心接口。 使用 Depends(dependencies.get_pagination_params) 自动从查询参数解析 page 和 size。 响应模型使用了泛型 PaginatedResponse[schemas.User] 来明确 items 的类型。 items, total crud.get_users_with_pagination(db, pagination) # 使用 Pydantic 模型的类方法创建标准化的分页响应 return schemas.PaginatedResponse.create( itemsitems, totaltotal, pagepagination.page, sizepagination.size )注意PaginatedResponse[schemas.User]这种泛型写法在 Pydantic V2 中支持良好。如果你使用 Pydantic V1可能需要稍微调整响应模型的定义方式或者直接使用response_modelschemas.PaginatedResponse并在文档中说明items的类型。4.4 启动服务在项目根目录fastapi-pagination-demo下运行以下命令启动开发服务器uvicorn app.main:app --reload --host 0.0.0.0 --port 8000app.main:app指定 FastAPI 应用实例的位置。--reload代码修改后自动重启服务器仅用于开发。--host 0.0.0.0允许本地网络访问如果仅本机访问可用127.0.0.1。--port 8000指定服务端口如果 8000 被占用可以换成其他端口如7860。启动成功后终端会显示Uvicorn running on http://0.0.0.0:8000。此时你可以通过浏览器访问http://127.0.0.1:8000/docs查看自动生成的交互式 API 文档Swagger UI。5. 功能测试与效果验证服务启动后我们通过几个步骤来验证分页功能是否正常工作。5.1 准备测试数据首先我们需要向数据库插入一些测试数据。你可以通过 API 文档界面手动创建也可以编写一个简单的脚本。这里我们通过交互式文档来操作。打开浏览器访问http://127.0.0.1:8000/docs。找到POST /users/接口点击 “Try it out”。在 Request body 中填入 JSON 数据例如{ username: john_doe, email: johnexample.com, full_name: John Doe }点击 “Execute”。如果成功响应码为 200并返回创建的用户信息。重复此步骤创建至少 15-20 个用户以便有足够的数据进行分页测试。5.2 测试分页接口现在测试核心的分页接口GET /users/。在 API 文档中找到GET /users/接口点击 “Try it out”。你会看到两个查询参数page和size它们已经有默认值1和10。直接点击 “Execute”调用默认的第一页page1, size10。观察响应items: 应返回一个包含最多 10 个用户对象的数组。total: 应等于你创建的用户总数。page: 应为 1。size: 应为 10。pages: 应根据total和size自动计算得出例如25条数据size10则 pages3。测试第二页将page参数改为2再次执行。items数组应返回第 11 到第 20 条数据如果存在。测试自定义页大小将size改为5page改为1执行。items数组应只包含 5 条数据pages总数会相应增加。测试边界情况请求一个超出范围的页码例如page999。此时items应该是一个空数组[]但total,page,size,pages等元信息依然正确返回。这是一种友好的处理方式。尝试传递size0或page0。由于我们在依赖项中设置了ge1FastAPI 会自动返回422 Unprocessable Entity错误并提示参数无效。这验证了参数验证在起作用。5.3 验证数据库查询效率可选对于性能敏感的应用需要确认分页查询没有性能问题。使用LIMIT/OFFSET在深度分页例如 page10000时性能会下降因为数据库需要扫描并跳过大量记录。如何观察你可以打开 SQLite 数据库查看工具或者为 SQLAlchemy 引擎开启echoTrue来查看执行的 SQL 语句。修改app/database.py中的create_engine部分engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False}, echoTrue # 开启 SQL 语句日志 )重启服务后在终端可以看到每次 API 调用实际执行的 SQL例如SELECT count(*) AS count_1 FROM users -- 计算总数的查询 SELECT * FROM users LIMIT ? OFFSET ? -- 获取当前页数据的查询这证实了我们的分页逻辑确实转换成了高效的数据库分页查询。6. 接口 API 与批量任务我们的分页接口本身就是一个标准的 RESTful API。前端可以通过简单的 HTTP GET 请求来调用。此外我们可以扩展这个模式来处理“批量任务”的概念例如后台导出所有分页数据。6.1 前端调用示例这里给出使用 JavaScript (Fetch API) 和 Python (requests 库) 调用分页接口的示例。JavaScript (在浏览器或 Node.js 中)async function fetchUsers(page 1, size 10) { const url http://127.0.0.1:8000/users/?page${page}size${size}; try { const response await fetch(url); if (!response.ok) { throw new Error(HTTP error! status: ${response.status}); } const data await response.json(); console.log(第 ${data.page} 页共 ${data.pages} 页); console.log(本页数据, data.items); console.log(总计${data.total} 条); return data; } catch (error) { console.error(获取用户列表失败:, error); } } // 调用示例 fetchUsers(1, 5);Pythonimport requests def fetch_users(page: int 1, size: int 10): url http://127.0.0.1:8000/users/ params {page: page, size: size} try: response requests.get(url, paramsparams, timeout10) response.raise_for_status() # 如果状态码不是200抛出异常 data response.json() print(f第 {data[page]} 页共 {data[pages]} 页) print(f本页数据量{len(data[items])}) print(f总计{data[total]} 条) return data except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None # 调用示例 fetch_users(page2, size5)6.2 模拟批量任务导出所有分页数据有时后端需要处理批量任务例如将一个查询结果的所有分页数据一次性处理如生成报表。虽然不推荐在前端循环调用但在后端服务内部或管理脚本中可以这样做。以下 Python 脚本演示了如何遍历所有分页安全地获取全部数据import requests import time def fetch_all_users(base_url: str, start_page: int 1, page_size: int 50, max_retries: int 3): 安全地获取所有用户数据。 base_url: API基础地址如 http://127.0.0.1:8000/users/ start_page: 起始页码 page_size: 每次请求获取的数据量 max_retries: 单次请求失败重试次数 all_users [] current_page start_page total_pages None while True: retries 0 success False while retries max_retries and not success: try: params {page: current_page, size: page_size} response requests.get(base_url, paramsparams, timeout30) response.raise_for_status() data response.json() success True except requests.exceptions.RequestException as e: retries 1 print(f请求第 {current_page} 页失败第 {retries} 次重试。错误: {e}) if retries max_retries: print(f第 {current_page} 页请求彻底失败停止获取。) return all_users time.sleep(2 ** retries) # 指数退避 if not success: break # 处理当前页数据 page_items data.get(items, []) all_users.extend(page_items) # 更新总页数通常第一页返回后就知道总页数了 if total_pages is None: total_pages data.get(pages, 1) print(f已获取第 {current_page}/{total_pages} 页累计 {len(all_users)} 条记录。) # 判断是否还有下一页 if current_page total_pages or not page_items: break current_page 1 print(f全部获取完成总计 {len(all_users)} 条用户数据。) return all_users # 使用示例 if __name__ __main__: all_users fetch_all_users(http://127.0.0.1:8000/users/, page_size20) # 接下来可以对 all_users 进行处理如写入文件、分析等关键点错误重试网络请求可能失败加入了简单的重试机制。指数退避重试等待时间逐渐增加避免对服务器造成压力。循环终止条件根据 API 返回的pages或items是否为空来判断是否结束。进度提示打印日志方便监控任务进度。7. 资源占用与性能观察对于一个分页 API性能瓶颈主要在于数据库查询而非 FastAPI 框架本身。以下是需要关注的性能要点和观察方法。7.1 数据库查询性能COUNT(*)查询get_users_with_pagination函数中db.query(func.count(...)).scalar()会触发一次全表计数。在数据量极大数百万时这个查询可能很慢。优化方案对于超大数据集可以考虑使用估算行数如 PostgreSQL 的pg_class.reltuples或者将总数缓存起来定期更新。OFFSET深度分页问题OFFSET 10000 LIMIT 10意味着数据库需要先扫描并跳过前 10000 条记录再取 10 条。随着OFFSET增大性能线性下降。优化方案游标分页使用WHERE id last_id LIMIT n代替OFFSET。这要求记录有唯一、递增的列如自增ID、创建时间戳。接口参数从page/size变为cursor上一页最后一条记录的ID和size。这能实现常数时间的翻页但无法直接跳转到任意页码。7.2 服务端资源观察内存占用主要取决于每页返回的数据量 (size)。确保size有上限我们设置了le100防止单次查询加载过多数据到内存。使用 Pydantic 模型序列化 ORM 对象也会产生内存开销但在合理页大小下通常不是问题。CPU 占用FastAPI 和 Pydantic 的序列化/反序列化非常高效。CPU 瓶颈更可能出现在复杂的数据库查询或业务逻辑上。网络 I/O响应体大小直接影响传输时间。对于包含大量文本或嵌套关系的列表可以考虑对响应进行压缩FastAPI 的GZipMiddleware或只返回必要字段。7.3 如何进行压力测试简易版你可以使用locust或wrk进行简单的压力测试观察接口在并发请求下的表现。使用requests进行简单并发测试Pythonimport concurrent.futures import requests import time BASE_URL http://127.0.0.1:8000/users/ def make_request(page): try: response requests.get(BASE_URL, params{page: page, size: 10}, timeout5) return response.status_code except Exception as e: return str(e) def stress_test(num_requests100, max_workers10): start_time time.time() with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 模拟请求不同的页码 futures [executor.submit(make_request, i % 10 1) for i in range(num_requests)] results [] for future in concurrent.futures.as_completed(futures): results.append(future.result()) end_time time.time() success sum(1 for r in results if r 200) print(f总请求数: {num_requests}) print(f成功请求: {success}) print(f总耗时: {end_time - start_time:.2f} 秒) print(f平均每秒请求数 (RPS): {num_requests / (end_time - start_time):.2f}) if __name__ __main__: stress_test(num_requests200, max_workers20)注意这只是一个简易测试在生产环境请使用专业的压测工具并在测试环境进行。8. 常见问题与排查方法在开发和部署分页接口时你可能会遇到以下问题。问题现象可能原因排查方式解决方案访问http://127.0.0.1:8000/docs无响应1. 服务未启动。2. 端口被占用。3. 防火墙/安全组阻止。1. 检查终端是否有Uvicorn running on日志。2. 运行netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用。3. 尝试用curl http://127.0.0.1:8000或浏览器直接访问根路径。1. 确保 uvicorn 命令正确执行。2. 更换端口如--port 7860。3. 检查防火墙设置开发时可暂时关闭。API 返回422 Unprocessable Entity请求参数不符合 Pydantic 模型验证规则。查看错误响应体通常会有detail字段指出哪个字段无效。例如[{loc: [query, size], msg: ensure this value is less than or equal to 100, ...}]。检查调用 API 时传递的参数。例如size不能大于100page不能小于1。确保前端传递的参数类型正确字符串还是数字。分页数据重复或丢失1. 数据库数据在分页查询间发生了增删。2. 排序不稳定。1. 检查两次查询间是否有插入或删除操作。2. 检查 SQL 查询是否缺少ORDER BY。如果顺序不固定OFFSET可能会指向不确定的位置。关键始终为分页查询添加ORDER BY子句且排序字段组合应能唯一确定顺序例如ORDER BY id DESC。这能保证分页结果的稳定性。COUNT(*)查询非常慢数据表行数巨大且没有合适的索引。使用数据库的EXPLAIN ANALYZE命令分析COUNT查询的执行计划。1. 考虑是否必须精确计数可用估算值替代。2. 在常用过滤条件字段上加索引。3. 定期将总数缓存到 Redis 或其他缓存中。深度分页大 offset响应慢OFFSET性能瓶颈。观察数据库监控或慢查询日志。1. 业务上限制最大可访问页码。2. 改用游标分页基于 ID 或时间戳。3. 使用覆盖索引优化查询。前端收到数据但无法渲染1. 响应格式与前端预期不符。2. Pydantic 模型序列化失败如包含不可JSON化的对象。1. 使用浏览器开发者工具或 Postman 查看 API 返回的实际 JSON 结构。2. 检查后端日志是否有序列化错误。1. 确保前端解析的是response.data.items而非response.data本身。2. 在 Pydantic 模型中使用orm_modeTrue并确保从数据库查询出的对象正确转换为 Pydantic 模型。复杂的字段如 datetime需要可 JSON 序列化。PaginatedResponse[schemas.User]类型错误Pydantic 版本或泛型支持问题。查看启动或请求时的错误日志。如果使用 Pydantic V1可以简化响应模型1. 在路由装饰器中直接使用response_modelschemas.PaginatedResponse。2. 在PaginatedResponse模型中将items: List[Any]改为items: List[schemas.User]但这会失去通用性。更推荐升级到 Pydantic V2。9. 最佳实践与使用建议基于以上实现和问题排查总结出以下最佳实践可以帮助你构建更健壮的分页系统。始终使用ORDER BY这是分页稳定性的基石。选择一个或一组能唯一确定顺序的字段进行排序如主键id或创建时间created_at。为分页参数设置合理边界page最小值应为 1。size必须设置最大值如 100防止恶意请求导致数据库过载。最小值通常为 1。提供默认值如 page1, size20提升 API 易用性。考虑游标分页以优化性能如果您的应用需要无限滚动或对深度分页性能要求高尽早设计基于游标Cursor的分页接口。参数可以是after_id和limit。谨慎处理COUNT查询在超大规模数据下精确COUNT代价高昂。评估业务是否真的需要精确总数。很多场景下“加载更多”按钮比显示“共 1000000 页”更友好。使用依赖注入管理分页参数如本文所示将page和size的解析与验证逻辑封装成依赖项 (get_pagination_params)使路由函数更简洁且便于统一修改验证规则。设计统一的响应格式PaginatedResponse模型让前端可以统一处理分页数据。可以在此基础上扩展例如加入has_next、has_prev布尔字段方便前端判断。为列表查询添加过滤和搜索真实场景的分页往往伴随过滤如statusactive和搜索如qjohn。将这些参数也设计为查询参数并在数据库查询中通过filter条件整合。编写清晰的 API 文档利用 FastAPI 自动生成的交互式文档并为每个查询参数添加description。这能极大降低前后端的沟通成本。进行自动化测试为分页接口编写单元测试和集成测试覆盖正常分页、边界值如最后一页、错误参数等场景。监控与告警在生产环境中监控分页接口的响应时间、错误率。如果COUNT查询或深度分页查询变慢应及时收到告警。10. 总结与下一步通过本文的实践我们完成了一个从零到一的 FastAPI 分页接口。它的核心价值在于提供了一套标准化、可复用、易于理解的后端分页解决方案。你不仅可以直接将代码集成到项目中更重要的是掌握了其设计思想依赖注入处理参数、Pydantic 模型定义响应、ORM 构建高效查询。最值得尝试的点快速集成将dependencies.py中的get_pagination_params和schemas.py中的PaginatedResponse复制到你的项目即可快速为任何列表接口添加分页功能。自动验证与文档FastAPI 基于类型提示和 Pydantic 自动生成的参数验证和交互式文档让前后端协作非常顺畅。最先应该验证的功能启动服务访问/docs测试分页接口的参数验证如输入size0或page-1。插入一批测试数据验证不同page和size组合下的返回结果是否正确特别是数据总数和总页数的计算。从前端或使用curl/Postman调用接口确保能正确解析响应中的items和分页元数据。最容易踩的坑忘记ORDER BY导致分页结果不稳定数据重复或丢失。size无上限被恶意请求传入size100000拖垮数据库。深度分页性能当数据量增长后原始的OFFSET方案需要优化。后续扩展方向异步化改造将sqlalchemy同步引擎替换为sqlalchemy.ext.asyncio异步引擎并使用asyncpg等异步驱动提升高并发下的性能。游标分页实现基于id或created_at实现cursor/next_token风格的分页解决深度分页性能问题。集成过滤与排序在分页依赖项中增加sort_by、sort_order、filter_xxx等参数实现更复杂的查询。缓存优化对COUNT查询结果或热点数据进行缓存减少数据库压力。接入前端框架编写 Vue/React 组件将分页响应数据与前端分页器组件如 Element UI Pagination、Ant Design Pagination绑定形成完整的前后端分页流程。建议将本文的核心代码块保存为本地模板在下次需要实现分页功能时它将成为你可靠的起点。
这次我们来看一个 FastAPI 实现前后端分页的实战项目。对于任何需要处理列表数据的 Web 应用分页都是核心功能它直接关系到用户体验和服务器性能。这个项目的重点不是概念多复杂而是如何用 FastAPI 快速、优雅地搭建一套可复用的分页接口并让前端能轻松对接。如果你关心如何在后端高效处理大数据集、如何设计标准化的分页响应格式、以及如何避免前端一次性加载过多数据导致的卡顿这篇文章可以直接收藏。我们将从零开始构建一个包含完整分页逻辑的 FastAPI 应用涵盖数据库查询、参数验证、响应模型设计以及前端调用示例。整个过程不依赖复杂的框架代码清晰易于集成到你的现有项目中。1. 核心能力速览能力项说明技术栈FastAPI (后端API框架) SQLAlchemy (ORM) Pydantic (数据验证)核心功能实现标准化的后端分页查询接口支持页码/页大小、排序、过滤等常见参数。接口设计RESTful 风格返回结构化的分页数据列表、总数、当前页、总页数等。性能考量通过数据库的LIMIT和OFFSET或更优的keyset pagination实现高效查询避免全表扫描。前端适配提供通用的 JSON 响应格式方便 Vue/React/Angular 等前端框架直接使用。启动方式通过uvicorn命令一键启动本地开发服务器。适合场景需要展示用户列表、商品列表、订单记录、日志数据等任何分页需求的 Web 应用后端开发。2. 适用场景与使用边界适合谁全栈开发者需要快速为前端提供分页数据接口。后端工程师希望构建一套标准、可维护的分页逻辑避免在每个接口重复造轮子。学习者想通过一个完整案例深入理解 FastAPI 的依赖注入、响应模型和数据库交互。能解决什么问题性能问题当数据库表数据量巨大时一次性查询所有数据会导致内存溢出、响应缓慢。分页接口只返回当前页所需的数据。体验问题前端无限滚动或页码切换需要结构化的分页信息如总条数、总页数来渲染分页器。标准化问题团队内不同接口的分页响应格式不统一增加前端对接成本。不适合什么场景实时流式数据如股票行情、聊天消息更适合使用 WebSocket 进行推送。极少量数据如果数据量固定且很少例如少于50条一次性返回可能更简单。需要复杂游标分页的场景本文主要基于页码-页大小模式对于深度分页优化如基于时间戳或ID的游标分页仅作原理性提及。使用边界与合规提醒确保分页查询涉及的数据查询符合隐私和数据安全法规避免通过分页参数恶意遍历非授权数据。在公开API中需要对分页参数如page_size设置合理的上限防止DoS攻击。3. 环境准备与前置条件在开始编码前请确保你的开发环境满足以下要求。这是一个通用清单你可以根据实际项目调整。操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。Python 版本Python 3.8 或更高版本。推荐使用 Python 3.10 以获得更好的类型提示支持。包管理工具pip通常随 Python 安装。核心依赖fastapi: 用于构建 API。uvicorn: 用于运行 ASGI 服务器。sqlalchemy: 用于数据库 ORM 操作。pydantic: 用于数据验证和设置管理。databases或asyncpg/aiomysql: 用于异步数据库驱动本文示例使用同步的sqlalchemy以简化但会说明异步方案。数据库SQLite用于演示、PostgreSQL 或 MySQL用于生产。可选工具pipenv或poetry: 用于虚拟环境和依赖管理。pgadmin/dbeaver: 数据库图形化管理工具。curl/Postman/Insomnia: 用于 API 测试。检查清单打开终端或 CMD/PowerShell运行python --version确认 Python 版本。运行pip --version确认 pip 可用。准备一个干净的项目目录例如fastapi-pagination-demo。4. 安装部署与启动方式我们将创建一个最小化的 FastAPI 应用来演示分页。首先在项目目录中安装依赖。4.1 创建虚拟环境与安装依赖建议使用虚拟环境隔离项目依赖。# 进入项目目录 cd fastapi-pagination-demo # 创建虚拟环境 (Windows) python -m venv venv # 激活虚拟环境 (Windows) venv\Scripts\activate # 创建虚拟环境 (macOS/Linux) python3 -m venv venv # 激活虚拟环境 (macOS/Linux) source venv/bin/activate虚拟环境激活后终端提示符前通常会出现(venv)标识。接着安装核心包pip install fastapi uvicorn sqlalchemy pydantic为了演示我们使用 SQLite 数据库它无需额外安装驱动。如果你计划使用 PostgreSQL可以安装asyncpg和databases。4.2 项目结构初始化创建以下文件和目录结构fastapi-pagination-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用实例和路由 │ ├── database.py # 数据库连接和会话管理 │ ├── models.py # SQLAlchemy 数据模型 │ ├── schemas.py # Pydantic 响应/请求模型 │ ├── crud.py # 增删改查操作包含分页逻辑 │ └── dependencies.py # 依赖项如分页参数依赖 ├── requirements.txt └── README.md4.3 核心代码实现下面我们分步骤填充核心文件。1. 数据库配置 (app/database.py)这里我们使用 SQLAlchemy 的同步引擎进行演示。对于生产环境强烈建议使用异步引擎 (asyncpgsqlalchemy.ext.asyncio)。from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # SQLite 数据库URL数据库文件将位于项目根目录的 sql_app.db SQLALCHEMY_DATABASE_URL sqlite:///./sql_app.db # 如果是 PostgreSQLURL类似 postgresql://user:passwordlocalhost/dbname # create_engine 的参数 connect_args 仅 SQLite 需要 engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) # 创建配置过的 SessionLocal 类 SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 创建 DeclarativeMeta 基类后续模型将继承它 Base declarative_base() # 依赖项获取数据库会话 def get_db(): db SessionLocal() try: yield db finally: db.close()2. 数据模型 (app/models.py)创建一个简单的User模型作为分页的数据源。from sqlalchemy import Column, Integer, String from .database import Base class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) username Column(String, uniqueTrue, indexTrue, nullableFalse) email Column(String, uniqueTrue, indexTrue, nullableFalse) full_name Column(String)3. Pydantic 模型 (app/schemas.py)定义请求和响应的数据结构。这是实现标准化分页响应的关键。from pydantic import BaseModel from typing import List, Optional, Any # 用户的基础模型 class UserBase(BaseModel): username: str email: str full_name: Optional[str] None class UserCreate(UserBase): pass class User(UserBase): id: int class Config: orm_mode True # 允许从 ORM 对象创建 Pydantic 模型 # --- 分页相关模型 --- # 通用的分页参数模型将由依赖项使用 class PaginationParams(BaseModel): page: int 1 size: int 10 class Config: # 允许通过查询参数如 ?page1size10传入 # 在依赖项中我们会用 Depends() 来接收而不是直接作为请求体 # 这里配置主要是为了文档和可能的其他用途 pass # 通用的分页响应模型 class PaginatedResponse(BaseModel): items: List[Any] # 当前页的数据列表 total: int # 数据总条数 page: int # 当前页码 size: int # 每页大小 pages: int # 总页数 classmethod def create(cls, items: List[Any], total: int, page: int, size: int): 便捷的创建方法自动计算总页数 pages (total size - 1) // size if size 0 else 0 return cls(itemsitems, totaltotal, pagepage, sizesize, pagespages)4. 分页参数依赖 (app/dependencies.py)我们将分页参数page和size提取为依赖项这样任何需要分页的接口都可以直接注入代码更清晰。from fastapi import Query from .schemas import PaginationParams def get_pagination_params( page: int Query(1, ge1, description页码从1开始), size: int Query(10, ge1, le100, description每页数量最大100) ) - PaginationParams: 分页参数依赖项。 从查询参数中获取 page 和 size并返回 PaginationParams 对象。 ge1 确保参数 1, le100 限制 size 最大为100防止过大查询。 return PaginationParams(pagepage, sizesize)5. 数据操作与分页逻辑 (app/crud.py)这里封装了用户相关的数据库操作核心是get_users分页函数。from sqlalchemy.orm import Session from . import models, schemas from sqlalchemy import func def create_user(db: Session, user: schemas.UserCreate): db_user models.User(**user.dict()) db.add(db_user) db.commit() db.refresh(db_user) return db_user def get_user(db: Session, user_id: int): return db.query(models.User).filter(models.User.id user_id).first() def get_users(db: Session, skip: int 0, limit: int 100): # 简单的 limit/offset 分页适用于数据量不大或深度分页不频繁的场景 return db.query(models.User).offset(skip).limit(limit).all() def get_users_with_pagination(db: Session, pagination: schemas.PaginationParams): 带完整分页信息的查询。 返回 (当前页的用户列表, 用户总数) # 计算 offset offset (pagination.page - 1) * pagination.size # 查询当前页数据 items db.query(models.User).offset(offset).limit(pagination.size).all() # 查询总条数这是一个单独的 count 查询在数据量大时需考虑性能 total db.query(func.count(models.User.id)).scalar() return items, total6. 主应用与路由 (app/main.py)将所有部分组合起来创建 FastAPI 应用并定义路由。from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from . import crud, models, schemas, dependencies from .database import engine, get_db # 创建数据库表仅演示生产环境请使用迁移工具如 Alembic models.Base.metadata.create_all(bindengine) app FastAPI(titleFastAPI 分页演示, version1.0.0) app.post(/users/, response_modelschemas.User) def create_user(user: schemas.UserCreate, db: Session Depends(get_db)): 创建新用户。 db_user crud.get_user_by_email(db, emailuser.email) # 需要实现 get_user_by_email if db_user: raise HTTPException(status_code400, detailEmail already registered) return crud.create_user(dbdb, useruser) app.get(/users/{user_id}, response_modelschemas.User) def read_user(user_id: int, db: Session Depends(get_db)): 根据ID获取单个用户。 db_user crud.get_user(db, user_iduser_id) if db_user is None: raise HTTPException(status_code404, detailUser not found) return db_user app.get(/users/, response_modelschemas.PaginatedResponse[schemas.User]) def read_users( pagination: schemas.PaginationParams Depends(dependencies.get_pagination_params), db: Session Depends(get_db) ): 获取用户列表带分页。 这是本文的核心接口。 使用 Depends(dependencies.get_pagination_params) 自动从查询参数解析 page 和 size。 响应模型使用了泛型 PaginatedResponse[schemas.User] 来明确 items 的类型。 items, total crud.get_users_with_pagination(db, pagination) # 使用 Pydantic 模型的类方法创建标准化的分页响应 return schemas.PaginatedResponse.create( itemsitems, totaltotal, pagepagination.page, sizepagination.size )注意PaginatedResponse[schemas.User]这种泛型写法在 Pydantic V2 中支持良好。如果你使用 Pydantic V1可能需要稍微调整响应模型的定义方式或者直接使用response_modelschemas.PaginatedResponse并在文档中说明items的类型。4.4 启动服务在项目根目录fastapi-pagination-demo下运行以下命令启动开发服务器uvicorn app.main:app --reload --host 0.0.0.0 --port 8000app.main:app指定 FastAPI 应用实例的位置。--reload代码修改后自动重启服务器仅用于开发。--host 0.0.0.0允许本地网络访问如果仅本机访问可用127.0.0.1。--port 8000指定服务端口如果 8000 被占用可以换成其他端口如7860。启动成功后终端会显示Uvicorn running on http://0.0.0.0:8000。此时你可以通过浏览器访问http://127.0.0.1:8000/docs查看自动生成的交互式 API 文档Swagger UI。5. 功能测试与效果验证服务启动后我们通过几个步骤来验证分页功能是否正常工作。5.1 准备测试数据首先我们需要向数据库插入一些测试数据。你可以通过 API 文档界面手动创建也可以编写一个简单的脚本。这里我们通过交互式文档来操作。打开浏览器访问http://127.0.0.1:8000/docs。找到POST /users/接口点击 “Try it out”。在 Request body 中填入 JSON 数据例如{ username: john_doe, email: johnexample.com, full_name: John Doe }点击 “Execute”。如果成功响应码为 200并返回创建的用户信息。重复此步骤创建至少 15-20 个用户以便有足够的数据进行分页测试。5.2 测试分页接口现在测试核心的分页接口GET /users/。在 API 文档中找到GET /users/接口点击 “Try it out”。你会看到两个查询参数page和size它们已经有默认值1和10。直接点击 “Execute”调用默认的第一页page1, size10。观察响应items: 应返回一个包含最多 10 个用户对象的数组。total: 应等于你创建的用户总数。page: 应为 1。size: 应为 10。pages: 应根据total和size自动计算得出例如25条数据size10则 pages3。测试第二页将page参数改为2再次执行。items数组应返回第 11 到第 20 条数据如果存在。测试自定义页大小将size改为5page改为1执行。items数组应只包含 5 条数据pages总数会相应增加。测试边界情况请求一个超出范围的页码例如page999。此时items应该是一个空数组[]但total,page,size,pages等元信息依然正确返回。这是一种友好的处理方式。尝试传递size0或page0。由于我们在依赖项中设置了ge1FastAPI 会自动返回422 Unprocessable Entity错误并提示参数无效。这验证了参数验证在起作用。5.3 验证数据库查询效率可选对于性能敏感的应用需要确认分页查询没有性能问题。使用LIMIT/OFFSET在深度分页例如 page10000时性能会下降因为数据库需要扫描并跳过大量记录。如何观察你可以打开 SQLite 数据库查看工具或者为 SQLAlchemy 引擎开启echoTrue来查看执行的 SQL 语句。修改app/database.py中的create_engine部分engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False}, echoTrue # 开启 SQL 语句日志 )重启服务后在终端可以看到每次 API 调用实际执行的 SQL例如SELECT count(*) AS count_1 FROM users -- 计算总数的查询 SELECT * FROM users LIMIT ? OFFSET ? -- 获取当前页数据的查询这证实了我们的分页逻辑确实转换成了高效的数据库分页查询。6. 接口 API 与批量任务我们的分页接口本身就是一个标准的 RESTful API。前端可以通过简单的 HTTP GET 请求来调用。此外我们可以扩展这个模式来处理“批量任务”的概念例如后台导出所有分页数据。6.1 前端调用示例这里给出使用 JavaScript (Fetch API) 和 Python (requests 库) 调用分页接口的示例。JavaScript (在浏览器或 Node.js 中)async function fetchUsers(page 1, size 10) { const url http://127.0.0.1:8000/users/?page${page}size${size}; try { const response await fetch(url); if (!response.ok) { throw new Error(HTTP error! status: ${response.status}); } const data await response.json(); console.log(第 ${data.page} 页共 ${data.pages} 页); console.log(本页数据, data.items); console.log(总计${data.total} 条); return data; } catch (error) { console.error(获取用户列表失败:, error); } } // 调用示例 fetchUsers(1, 5);Pythonimport requests def fetch_users(page: int 1, size: int 10): url http://127.0.0.1:8000/users/ params {page: page, size: size} try: response requests.get(url, paramsparams, timeout10) response.raise_for_status() # 如果状态码不是200抛出异常 data response.json() print(f第 {data[page]} 页共 {data[pages]} 页) print(f本页数据量{len(data[items])}) print(f总计{data[total]} 条) return data except requests.exceptions.RequestException as e: print(f请求失败: {e}) return None # 调用示例 fetch_users(page2, size5)6.2 模拟批量任务导出所有分页数据有时后端需要处理批量任务例如将一个查询结果的所有分页数据一次性处理如生成报表。虽然不推荐在前端循环调用但在后端服务内部或管理脚本中可以这样做。以下 Python 脚本演示了如何遍历所有分页安全地获取全部数据import requests import time def fetch_all_users(base_url: str, start_page: int 1, page_size: int 50, max_retries: int 3): 安全地获取所有用户数据。 base_url: API基础地址如 http://127.0.0.1:8000/users/ start_page: 起始页码 page_size: 每次请求获取的数据量 max_retries: 单次请求失败重试次数 all_users [] current_page start_page total_pages None while True: retries 0 success False while retries max_retries and not success: try: params {page: current_page, size: page_size} response requests.get(base_url, paramsparams, timeout30) response.raise_for_status() data response.json() success True except requests.exceptions.RequestException as e: retries 1 print(f请求第 {current_page} 页失败第 {retries} 次重试。错误: {e}) if retries max_retries: print(f第 {current_page} 页请求彻底失败停止获取。) return all_users time.sleep(2 ** retries) # 指数退避 if not success: break # 处理当前页数据 page_items data.get(items, []) all_users.extend(page_items) # 更新总页数通常第一页返回后就知道总页数了 if total_pages is None: total_pages data.get(pages, 1) print(f已获取第 {current_page}/{total_pages} 页累计 {len(all_users)} 条记录。) # 判断是否还有下一页 if current_page total_pages or not page_items: break current_page 1 print(f全部获取完成总计 {len(all_users)} 条用户数据。) return all_users # 使用示例 if __name__ __main__: all_users fetch_all_users(http://127.0.0.1:8000/users/, page_size20) # 接下来可以对 all_users 进行处理如写入文件、分析等关键点错误重试网络请求可能失败加入了简单的重试机制。指数退避重试等待时间逐渐增加避免对服务器造成压力。循环终止条件根据 API 返回的pages或items是否为空来判断是否结束。进度提示打印日志方便监控任务进度。7. 资源占用与性能观察对于一个分页 API性能瓶颈主要在于数据库查询而非 FastAPI 框架本身。以下是需要关注的性能要点和观察方法。7.1 数据库查询性能COUNT(*)查询get_users_with_pagination函数中db.query(func.count(...)).scalar()会触发一次全表计数。在数据量极大数百万时这个查询可能很慢。优化方案对于超大数据集可以考虑使用估算行数如 PostgreSQL 的pg_class.reltuples或者将总数缓存起来定期更新。OFFSET深度分页问题OFFSET 10000 LIMIT 10意味着数据库需要先扫描并跳过前 10000 条记录再取 10 条。随着OFFSET增大性能线性下降。优化方案游标分页使用WHERE id last_id LIMIT n代替OFFSET。这要求记录有唯一、递增的列如自增ID、创建时间戳。接口参数从page/size变为cursor上一页最后一条记录的ID和size。这能实现常数时间的翻页但无法直接跳转到任意页码。7.2 服务端资源观察内存占用主要取决于每页返回的数据量 (size)。确保size有上限我们设置了le100防止单次查询加载过多数据到内存。使用 Pydantic 模型序列化 ORM 对象也会产生内存开销但在合理页大小下通常不是问题。CPU 占用FastAPI 和 Pydantic 的序列化/反序列化非常高效。CPU 瓶颈更可能出现在复杂的数据库查询或业务逻辑上。网络 I/O响应体大小直接影响传输时间。对于包含大量文本或嵌套关系的列表可以考虑对响应进行压缩FastAPI 的GZipMiddleware或只返回必要字段。7.3 如何进行压力测试简易版你可以使用locust或wrk进行简单的压力测试观察接口在并发请求下的表现。使用requests进行简单并发测试Pythonimport concurrent.futures import requests import time BASE_URL http://127.0.0.1:8000/users/ def make_request(page): try: response requests.get(BASE_URL, params{page: page, size: 10}, timeout5) return response.status_code except Exception as e: return str(e) def stress_test(num_requests100, max_workers10): start_time time.time() with concurrent.futures.ThreadPoolExecutor(max_workersmax_workers) as executor: # 模拟请求不同的页码 futures [executor.submit(make_request, i % 10 1) for i in range(num_requests)] results [] for future in concurrent.futures.as_completed(futures): results.append(future.result()) end_time time.time() success sum(1 for r in results if r 200) print(f总请求数: {num_requests}) print(f成功请求: {success}) print(f总耗时: {end_time - start_time:.2f} 秒) print(f平均每秒请求数 (RPS): {num_requests / (end_time - start_time):.2f}) if __name__ __main__: stress_test(num_requests200, max_workers20)注意这只是一个简易测试在生产环境请使用专业的压测工具并在测试环境进行。8. 常见问题与排查方法在开发和部署分页接口时你可能会遇到以下问题。问题现象可能原因排查方式解决方案访问http://127.0.0.1:8000/docs无响应1. 服务未启动。2. 端口被占用。3. 防火墙/安全组阻止。1. 检查终端是否有Uvicorn running on日志。2. 运行netstat -ano | findstr :8000(Win) 或lsof -i:8000(Mac/Linux) 查看端口占用。3. 尝试用curl http://127.0.0.1:8000或浏览器直接访问根路径。1. 确保 uvicorn 命令正确执行。2. 更换端口如--port 7860。3. 检查防火墙设置开发时可暂时关闭。API 返回422 Unprocessable Entity请求参数不符合 Pydantic 模型验证规则。查看错误响应体通常会有detail字段指出哪个字段无效。例如[{loc: [query, size], msg: ensure this value is less than or equal to 100, ...}]。检查调用 API 时传递的参数。例如size不能大于100page不能小于1。确保前端传递的参数类型正确字符串还是数字。分页数据重复或丢失1. 数据库数据在分页查询间发生了增删。2. 排序不稳定。1. 检查两次查询间是否有插入或删除操作。2. 检查 SQL 查询是否缺少ORDER BY。如果顺序不固定OFFSET可能会指向不确定的位置。关键始终为分页查询添加ORDER BY子句且排序字段组合应能唯一确定顺序例如ORDER BY id DESC。这能保证分页结果的稳定性。COUNT(*)查询非常慢数据表行数巨大且没有合适的索引。使用数据库的EXPLAIN ANALYZE命令分析COUNT查询的执行计划。1. 考虑是否必须精确计数可用估算值替代。2. 在常用过滤条件字段上加索引。3. 定期将总数缓存到 Redis 或其他缓存中。深度分页大 offset响应慢OFFSET性能瓶颈。观察数据库监控或慢查询日志。1. 业务上限制最大可访问页码。2. 改用游标分页基于 ID 或时间戳。3. 使用覆盖索引优化查询。前端收到数据但无法渲染1. 响应格式与前端预期不符。2. Pydantic 模型序列化失败如包含不可JSON化的对象。1. 使用浏览器开发者工具或 Postman 查看 API 返回的实际 JSON 结构。2. 检查后端日志是否有序列化错误。1. 确保前端解析的是response.data.items而非response.data本身。2. 在 Pydantic 模型中使用orm_modeTrue并确保从数据库查询出的对象正确转换为 Pydantic 模型。复杂的字段如 datetime需要可 JSON 序列化。PaginatedResponse[schemas.User]类型错误Pydantic 版本或泛型支持问题。查看启动或请求时的错误日志。如果使用 Pydantic V1可以简化响应模型1. 在路由装饰器中直接使用response_modelschemas.PaginatedResponse。2. 在PaginatedResponse模型中将items: List[Any]改为items: List[schemas.User]但这会失去通用性。更推荐升级到 Pydantic V2。9. 最佳实践与使用建议基于以上实现和问题排查总结出以下最佳实践可以帮助你构建更健壮的分页系统。始终使用ORDER BY这是分页稳定性的基石。选择一个或一组能唯一确定顺序的字段进行排序如主键id或创建时间created_at。为分页参数设置合理边界page最小值应为 1。size必须设置最大值如 100防止恶意请求导致数据库过载。最小值通常为 1。提供默认值如 page1, size20提升 API 易用性。考虑游标分页以优化性能如果您的应用需要无限滚动或对深度分页性能要求高尽早设计基于游标Cursor的分页接口。参数可以是after_id和limit。谨慎处理COUNT查询在超大规模数据下精确COUNT代价高昂。评估业务是否真的需要精确总数。很多场景下“加载更多”按钮比显示“共 1000000 页”更友好。使用依赖注入管理分页参数如本文所示将page和size的解析与验证逻辑封装成依赖项 (get_pagination_params)使路由函数更简洁且便于统一修改验证规则。设计统一的响应格式PaginatedResponse模型让前端可以统一处理分页数据。可以在此基础上扩展例如加入has_next、has_prev布尔字段方便前端判断。为列表查询添加过滤和搜索真实场景的分页往往伴随过滤如statusactive和搜索如qjohn。将这些参数也设计为查询参数并在数据库查询中通过filter条件整合。编写清晰的 API 文档利用 FastAPI 自动生成的交互式文档并为每个查询参数添加description。这能极大降低前后端的沟通成本。进行自动化测试为分页接口编写单元测试和集成测试覆盖正常分页、边界值如最后一页、错误参数等场景。监控与告警在生产环境中监控分页接口的响应时间、错误率。如果COUNT查询或深度分页查询变慢应及时收到告警。10. 总结与下一步通过本文的实践我们完成了一个从零到一的 FastAPI 分页接口。它的核心价值在于提供了一套标准化、可复用、易于理解的后端分页解决方案。你不仅可以直接将代码集成到项目中更重要的是掌握了其设计思想依赖注入处理参数、Pydantic 模型定义响应、ORM 构建高效查询。最值得尝试的点快速集成将dependencies.py中的get_pagination_params和schemas.py中的PaginatedResponse复制到你的项目即可快速为任何列表接口添加分页功能。自动验证与文档FastAPI 基于类型提示和 Pydantic 自动生成的参数验证和交互式文档让前后端协作非常顺畅。最先应该验证的功能启动服务访问/docs测试分页接口的参数验证如输入size0或page-1。插入一批测试数据验证不同page和size组合下的返回结果是否正确特别是数据总数和总页数的计算。从前端或使用curl/Postman调用接口确保能正确解析响应中的items和分页元数据。最容易踩的坑忘记ORDER BY导致分页结果不稳定数据重复或丢失。size无上限被恶意请求传入size100000拖垮数据库。深度分页性能当数据量增长后原始的OFFSET方案需要优化。后续扩展方向异步化改造将sqlalchemy同步引擎替换为sqlalchemy.ext.asyncio异步引擎并使用asyncpg等异步驱动提升高并发下的性能。游标分页实现基于id或created_at实现cursor/next_token风格的分页解决深度分页性能问题。集成过滤与排序在分页依赖项中增加sort_by、sort_order、filter_xxx等参数实现更复杂的查询。缓存优化对COUNT查询结果或热点数据进行缓存减少数据库压力。接入前端框架编写 Vue/React 组件将分页响应数据与前端分页器组件如 Element UI Pagination、Ant Design Pagination绑定形成完整的前后端分页流程。建议将本文的核心代码块保存为本地模板在下次需要实现分页功能时它将成为你可靠的起点。
