FastAPI 分页模块实现详解

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，必须大于等于1
page_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
}