41｜分页、排序与统一响应格式

列表接口一次性返回所有数据，前端渲染卡、网络传输慢、数据库压力大。分页把大数据集切分成小块，每次只传一页。统一响应格式让前端能一致地处理成功和错误，不需要为每个接口写不同的解析逻辑。

一、分页

limit/offset 分页

最常用：

@app.get("/servers")
async def list_servers(
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: Session = Depends(get_db),
):
    total = db.execute(select(func.count()).select_from(Server)).scalar()
    items = db.execute(
        select(Server).offset(skip).limit(limit)
    ).scalars().all()

    return {
        "items": items,
        "total": total,
        "skip": skip,
        "limit": limit,
    }

请求示例：

GET /servers?skip=0&limit=20     # 第 1 页
GET /servers?skip=20&limit=20    # 第 2 页
GET /servers?skip=40&limit=20    # 第 3 页

limit 必须设上限，防止 limit=999999 拖垮数据库。

游标分页

数据量大时，offset 越大性能越差——数据库要先扫描过前 offset 条再取数据。游标分页用上一页最后一条的排序字段值作为起点：

@app.get("/servers")
async def list_servers_cursor(
    cursor: str | None = None,
    limit: int = Query(20, ge=1, le=100),
    db: Session = Depends(get_db),
):
    stmt = select(Server).order_by(Server.id).limit(limit + 1)

    if cursor:
        stmt = stmt.where(Server.id > cursor)

    items = db.execute(stmt).scalars().all()
    has_more = len(items) > limit
    items = items[:limit]

    return {
        "items": items,
        "next_cursor": str(items[-1].id) if items and has_more else None,
        "has_more": has_more,
    }

游标分页的优点：性能稳定，适合大数据量。缺点：不能跳页（没有"第 5 页"的概念），需要稳定的排序字段。

二、排序

from sqlalchemy import asc, desc

@app.get("/servers")
async def list_servers(
    sort: str = "-created_at",
    db: Session = Depends(get_db),
):
    column_name = sort.lstrip("-")
    direction = desc if sort.startswith("-") else asc

    column = getattr(Server, column_name, Server.created_at)
    stmt = select(Server).order_by(direction(column))
    ...

参数值	含义
created_at	按创建时间升序
-created_at	按创建时间降序
hostname	按主机名升序

不要直接把用户输入映射到列名，需要白名单校验：

ALLOWED_SORT = {"created_at", "hostname", "status"}

if column_name not in ALLOWED_SORT:
    raise HTTPException(status_code=400, detail=f"不允许按 {column_name} 排序")

三、统一响应格式

生产环境通常要求所有接口返回统一的包装结构：

from typing import Generic, TypeVar
from pydantic import BaseModel
from pydantic.generics import GenericModel

T = TypeVar("T")

class ResponseWrapper(GenericModel, Generic[T]):
    code: int = 0
    message: str = "success"
    data: T | None = None

自定义异常处理器

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ResponseWrapper(
            code=exc.status_code,
            message=exc.detail,
            data=None,
        ).dict(),
    )

@app.exception_handler(Exception)
async def generic_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content=ResponseWrapper(
            code=500,
            message="服务器内部错误",
            data=None,
        ).dict(),
    )

分页响应模型

class PaginatedResponse(GenericModel, Generic[T]):
    code: int = 0
    message: str = "success"
    data: list[T]
    total: int
    skip: int
    limit: int

四、常见错误

limit 无上限

# 危险
@app.get("/servers")
async def list_servers(limit: int = 100):
    ...

# 正确：限制最大值
limit: int = Query(20, ge=1, le=100)

排序字段注入

# 危险：用户传入任意列名
stmt = select(Server).order_by(text(sort))

# 攻击：sort = "1; DROP TABLE servers;--"

# 正确：白名单校验
if sort.lstrip("-") not in ALLOWED_COLUMNS:
    raise HTTPException(status_code=400)

响应格式不统一

# 错误：有的接口直接返回列表，有的包装成对象
@app.get("/a")
async def a(): return [1, 2, 3]

@app.get("/b")
async def b(): return {"data": [1, 2, 3]}

# 正确：统一包装
return ResponseWrapper(data=[1, 2, 3])