FastAPI 核心机制：分页参数的实现与最佳实践

当前位置:

首页 > Python基础教程 >

FastAPI 核心机制：分页参数的实现与最佳实践

分页的基本概念
分页是将大量数据分割成多个小块（即“页”），以便用户或系统可以逐步加载和处理这些数据。在Web应用中，分页通常用于处理数据库查询结果、API响应等场景。常见的分页参数包括：

page：当前页码。
page_size：每页显示的数据条数。
total_pages：总页数。
2. FastAPI中的分页实现
在FastAPI中，分页可以通过查询参数来实现。以下是一个简单的示例，展示了如何在FastAPI中实现分页功能。

PYTHON

from fastapi import FastAPI, Query
from typing import List, Optional

app = FastAPI()

# 模拟数据库数据
fake_items_db = [{"item_name": f"Item {i}"} for i in range(100)]


@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0)):
    start = (page - 1) * page_size
    end = start + page_size
    items = fake_items_db[start:end]
    total_items = len(fake_items_db)
    total_pages = (total_items + page_size - 1) // page_size
    return {
        "items": items,
        "page": page,
        "page_size": page_size,
        "total_items": total_items,
        "total_pages": total_pages,
    }

在这个示例中，我们定义了两个查询参数page和page_size，并通过计算start和end
来获取当前页的数据。我们还计算了总页数total_pages，并将其包含在响应中。

分页参数的最佳实践
3.1 参数验证
为了确保分页参数的有效性，我们需要对page和page_size进行验证。FastAPI提供了Query参数验证功能，可以轻松实现这一点。

PYTHON

from fastapi import Query


@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100)):
    # 分页逻辑
    pass

在这个示例中，我们使用gt（大于）和le（小于等于）来限制page和page_size的取值范围。如果用户提供的参数不符合要求，FastAPI会自动返回422
Validation Error。

3.2 默认值设置
为分页参数设置合理的默认值可以提升用户体验。例如，将page_size的默认值设置为10或20，可以避免用户一次性加载过多数据。

PYTHON

@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100)):
    # 分页逻辑
    pass

3.3 总页数计算
总页数的计算公式为：

PYTHON
total_pages = (total_items + page_size - 1) // page_size
这个公式确保了总页数的准确性，即使total_items不能被page_size整除。

数据库查询中的分页
在实际应用中，分页通常与数据库查询结合使用。以下是一个使用SQLAlchemy进行分页查询的示例。

PYTHON

from sqlalchemy.orm import Session
from fastapi import Depends
from .database import SessionLocal


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100),
                     db: Session = Depends(get_db)):
    start = (page - 1) * page_size
    items = db.query(Item).offset(start).limit(page_size).all()
    total_items = db.query(Item).count()
    total_pages = (total_items + page_size - 1) // page_size
    return {
        "items": items,
        "page": page,
        "page_size": page_size,
        "total_items": total_items,
        "total_pages": total_pages,
    }

在这个示例中，我们使用offset和limit来实现分页查询，并通过count方法获取总数据条数。

分页的安全性
5.1 避免SQL注入
在使用原始SQL查询时，必须注意避免SQL注入攻击。SQLAlchemy等ORM框架已经内置了防止SQL注入的机制，但在使用原始SQL时，仍需谨慎。

PYTHON

from sqlalchemy.sql import text


@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100),
                     db: Session = Depends(get_db)):
    start = (page - 1) * page_size
    query = text("SELECT * FROM items LIMIT :limit OFFSET :offset")
    items = db.execute(query, {"limit": page_size, "offset": start}).fetchall()
    total_items = db.execute(text("SELECT COUNT(*) FROM items")).scalar()
    total_pages = (total_items + page_size - 1) // page_size
    return {
        "items": items,
        "page": page,
        "page_size": page_size,
        "total_items": total_items,
        "total_pages": total_pages,
    }

在这个示例中，我们使用参数化查询来避免SQL注入。

5.2 数据隐私
在处理敏感数据时，确保分页查询不会泄露隐私信息。例如，避免在分页查询中返回未授权的数据。

性能优化
6.1 索引优化
在数据库查询中，为分页字段（如id、created_at等）创建索引可以显著提升查询性能。

SQL
CREATE INDEX idx_items_created_at ON items (created_at);
6.2 缓存
对于频繁访问的分页数据，可以使用缓存机制（如Redis）来减少数据库查询次数。

PYTHON

from fastapi_cache import FastAPICache
from fastapi_cache.decorator import cache


@app.get("/items/")
@cache(expire=60)
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100)):
    # 分页逻辑
    pass

在这个示例中，我们使用fastapi-cache库来缓存分页查询结果，缓存有效期为60秒。

常见错误及解决方案
7.1 422 Validation Error
当分页参数不符合验证规则时，FastAPI会返回422 Validation Error。解决方案是确保分页参数的取值范围正确，并在API文档中明确说明。

PYTHON

@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100)):
    # 分页逻辑
    pass
7.2 500 Internal Server Error

当数据库查询失败或分页逻辑出现错误时，可能会返回500 Internal Server Error。解决方案是捕获异常并返回友好的错误信息。

PYTHON

from fastapi import HTTPException


@app.get("/items/")
async def read_items(page: int = Query(1, gt=0), page_size: int = Query(10, gt=0, le=100),
                     db: Session = Depends(get_db)):
    try:
        start = (page - 1) * page_size
        items = db.query(Item).offset(start).limit(page_size).all()
        total_items = db.query(Item).count()
        total_pages = (total_items + page_size - 1) // page_size
        return {
            "items": items,
            "page": page,
            "page_size": page_size,
            "total_items": total_items,
            "total_pages": total_pages,
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

在这个示例中，我们捕获了所有异常，并返回500 Internal Server Error。

课后Quiz
如何避免SQL注入攻击？

使用参数化查询。
避免拼接SQL语句。
使用ORM框架。
如何优化分页查询的性能？

为分页字段创建索引。
使用缓存机制。
减少查询返回的字段数量。
如何处理分页参数无效的情况？

使用FastAPI的Query参数验证功能。
返回422 Validation Error。
在API文档中明确说明参数要求。
常见报错解决方案：

422 Validation Error：检查分页参数的取值范围，确保符合验证规则。
500 Internal Server Error：捕获异常并返回友好的错误信息，检查数据库查询逻辑。
404 Not Found：确保分页参数不会导致查询结果为空，处理边界情况。

来源：https://blog.cmdragon.cn/posts/6a3cba67a72d/

栏目列表