1. 简介
本文详细介绍了一个基于 FastAPI 框架的通用分页处理模块的实现。该模块提供了标准的分页参数处理、数据切片和响应格式化功能,可以轻松地集成到任何 FastAPI 项目中。
2. 代码实现
2.1 导入必要的模块
首先,我们需要导入所需的模块:
from typing import Optional, List, Dict, Any
from fastapi import FastAPI, Query, HTTPException
from pydantic import BaseModel
from math import ceil
这些导入包括:
typing
: 提供类型提示支持fastapi
: FastAPI 框架的核心组件pydantic
: 用于数据验证的模型类math.ceil
: 用于向上取整计算总页数
2.2 分页参数模型
定义分页请求参数的数据模型:
class PaginationParams(BaseModel):"""分页参数模型Attributes:page: 当前页码,从1开始page_size: 每页数据条数total: 数据总条数"""page: int = Query(1, ge=1, description="当前页码,从1开始")page_size: int = Query(10, ge=1, le=100, description="每页数据条数,1-100之间")total: Optional[int] = None
这个模型定义了:
page
: 当前页码,默认为1,必须大于等于1page_size
: 每页数据条数,默认为10,范围在1-100之间total
: 可选的总数据条数
2.3 分页响应模型
定义标准的分页响应格式:
class PaginatedResponse(BaseModel):"""分页响应模型Attributes:items: 分页数据列表total: 数据总条数page: 当前页码page_size: 每页数据条数total_pages: 总页数has_next: 是否有下一页has_prev: 是否有上一页"""items: List[Dict[str, Any]]total: intpage: intpage_size: inttotal_pages: inthas_next: boolhas_prev: bool
响应模型包含:
items
: 当前页的数据列表total
: 数据总条数page
: 当前页码page_size
: 每页条数total_pages
: 总页数has_next
: 是否有下一页has_prev
: 是否有上一页
2.4 分页核心函数
实现分页处理的核心逻辑:
def paginate(items: List[Any],pagination: PaginationParams
) -> PaginatedResponse:"""通用分页函数Args:items: 需要分页的数据列表pagination: 分页参数对象Returns:PaginatedResponse: 分页后的数据响应对象Raises:HTTPException: 当页码超出范围时抛出异常"""# 计算总条数total = len(items)# 计算总页数total_pages = ceil(total / pagination.page_size)# 验证页码是否有效if pagination.page > total_pages and total > 0:raise HTTPException(status_code=404, detail="Page not found")# 计算当前页的数据切片start = (pagination.page - 1) * pagination.page_sizeend = start + pagination.page_size# 获取当前页数据current_items = items[start:end]# 构建分页响应return PaginatedResponse(items=current_items,total=total,page=pagination.page,page_size=pagination.page_size,total_pages=total_pages,has_next=pagination.page < total_pages,has_prev=pagination.page > 1)
核心函数实现了:
- 计算数据总条数
- 计算总页数
- 验证页码有效性
- 计算数据切片范围
- 获取当前页数据
- 构建标准响应
3. 使用示例
3.1 基本用法
from fastapi import FastAPI, Dependsapp = FastAPI()# 示例数据
items = [{"id": i, "name": f"Item {i}"} for i in range(100)]@app.get("/items/", response_model=PaginatedResponse)
async def get_items(pagination: PaginationParams = Depends()):return paginate(items, pagination)
3.2 API 调用示例
# 获取第一页,每页10条数据
GET /items/?page=1&page_size=10# 获取第二页,每页20条数据
GET /items/?page=2&page_size=20
3.3 响应示例
{"items": [{"id": 0, "name": "Item 0"},{"id": 1, "name": "Item 1"},// ... 更多数据],"total": 100,"page": 1,"page_size": 10,"total_pages": 10,"has_next": true,"has_prev": false
}
4. 特点和优势
-
类型安全
- 使用 Python 类型注解
- 使用 Pydantic 模型进行数据验证
- IDE 友好,提供代码补全支持
-
参数验证
- 自动验证页码和每页条数
- 防止无效的分页参数
- 提供清晰的错误信息
-
标准响应
- 统一的响应格式
- 包含分页元数据
- 便于前端处理
-
异常处理
- 优雅处理无效页码
- 返回标准的 HTTP 错误码
- 提供清晰的错误信息
5. 注意事项
-
页码计数
- 页码从 1 开始计数
- 无效页码会返回 404 错误
-
数据限制
- 每页条数限制在 1-100 之间
- 可以根据需要调整限制范围
-
性能考虑
- 适用于内存中的数据列表
- 对于数据库查询,建议使用数据库级别的分页
6. 总结
这个分页模块提供了一个完整的解决方案,可以轻松处理 FastAPI 应用中的分页需求。它的设计注重:
- 代码的可读性和可维护性
- 类型安全和参数验证
- 标准化的响应格式
- 良好的错误处理
通过使用这个模块,可以大大简化 API 开发中的分页实现,提高开发效率和代码质量。