请求参数、响应模型与状态码

第 4 章 · 请求参数、响应模型与状态码

本章目标：掌握 FastAPI Path、Query、Body 参数声明与自动校验；使用 response_model 过滤响应字段、控制序列化；正确设置 status_code（201 创建、204 无内容等）；用 HTTPException 与自定义异常处理返回标准错误体；完善 svc-demo 商品 CRUD 的 REST 语义；对照 Django DRF 与 Flask 的错误处理方式。

学时建议：4～5 小时（含 1.5 小时跟练）

前置：完成 fastapi-web ch03（Pydantic 模型）；熟悉 HTTP 状态码含义。

4.1 场景说明：RESTful 商品 API 契约

svc-demo 需在 api.example.com 提供完整商品 CRUD，前端要求：

本章聚焦参数绑定与响应契约；数据持久化在 ch05 接入数据库。

4.2 Path 路径参数

from fastapi import APIRouter, Path

router = APIRouter(prefix="/api/v1/products", tags=["商品"])


@router.get("/{product_id}")
async def get_product(
    product_id: int = Path(..., gt=0, description="商品 ID，正整数"),
):
    ...

多个路径参数：

@router.get("/by-category/{category_slug}/{product_id}")
async def get_by_category(
    category_slug: str = Path(..., pattern=r"^[a-z0-9-]+$"),
    product_id: int = Path(..., gt=0),
):
    return {"category_slug": category_slug, "product_id": product_id}

顺序规则：无默认值的 Path/Query 必须在有默认值参数之前；Body 模型通常放在最后。

4.3 Query 查询参数

from enum import Enum
from fastapi import Query

from app.dependencies import Pagination, get_pagination


class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"


@router.get("/")
async def list_products(
    pagination: Pagination = Depends(get_pagination),
    q: str | None = Query(None, min_length=1, max_length=64, description="关键词"),
    category_id: int | None = Query(None, gt=0),
    is_active: bool | None = Query(None),
    sort: SortOrder = Query(SortOrder.desc, description="排序方向"),
):
    items = ProductMemoryRepo.list_all()
    if q:
        items = [p for p in items if q.lower() in p.name.lower()]
    if category_id is not None:
        items = [p for p in items if p.category_id == category_id]
  # 分页切片
    start = pagination.offset
    end = start + pagination.limit
    return ProductListResponse(total=len(items), items=items[start:end])

4.4 Body 请求体

4.4.1 单模型 Body

from fastapi import Body

from app.schemas.product import ProductCreate, ProductRead, ProductUpdate


@router.post("/", response_model=ProductRead, status_code=201)
async def create_product(payload: ProductCreate):
    """创建商品 — 请求体自动绑定并校验"""
    return ProductMemoryRepo.create(payload)


@router.patch("/{product_id}", response_model=ProductRead)
async def update_product(
    product_id: int = Path(..., gt=0),
    payload: ProductUpdate = Body(...),
):
    updated = ProductMemoryRepo.update(product_id, payload)
    if not updated:
        raise HTTPException(status_code=404, detail="商品不存在")
    return updated

ProductUpdate 示例（ch03 练习）：

# app/schemas/product.py
from typing import Optional
from decimal import Decimal
from pydantic import BaseModel, Field


class ProductUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=128)
    price: Optional[Decimal] = Field(None, ge=0)
    stock: Optional[int] = Field(None, ge=0)
    is_active: Optional[bool] = None

4.5 Response Model 响应模型

@router.get("/", response_model=ProductListResponse)
async def list_products(...):
    ...

response_model 的作用：

排除 None 字段：

@router.get("/{product_id}", response_model=ProductDetail, response_model_exclude_none=True)
async def get_product(product_id: int):
    ...

4.6 status_code 状态码

from fastapi import status


@router.post("/", response_model=ProductRead, status_code=status.HTTP_201_CREATED)
async def create_product(payload: ProductCreate):
    return ProductMemoryRepo.create(payload)