FastAPI 分页模块实现详解

news2025/3/12 11:26:52

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: int
    page: int
    page_size: int
    total_pages: int
    has_next: bool
    has_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_size
    end = 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, Depends

app = 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
}