FastAPIrequest-body-handling与响应模型

📂 所属阶段：第一阶段 — 快速筑基（基础篇）
🔗 相关章节：path-query-parameters · response-handling-status-codes · pydantic-guide

目录

request-body-handling概述

请求体是Web API接收客户端数据的心脏。FastAPI 与 Pydantic 2.x 联手，为你提供了“自动验证 + 自动文档 + 类型转换”的三合一体验，彻底告别手动解析 JSON、手写校验逻辑的重复劳动。

核心处理流程

  1. 用Pydantic模型定义数据契约：字段类型、是否必填、约束规则一目了然
  2. FastAPI自动解析并验证：从HTTP请求中提取数据，转换成正确的Python类型并校验
  3. 友好的错误反馈：校验不通过时返回结构化的错误信息，便于客户端快速定位
  4. OpenAPI文档自动生成：模型中的约束直接映射为交互式文档里的示例与说明

Pydantic请求模型

基础模型与字段约束

最常见的做法是继承 BaseModel，再配合 Field 添加详细约束，像搭积木一样构建请求体。

from fastapi import FastAPI
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
import re

app = FastAPI()

class ProductCreate(BaseModel):
    """产品创建请求模型"""
    # 必填字段，附带最小/最大长度约束
    name: str = Field(..., min_length=3, max_length=100, description="产品名称")
    # 可选字段，可以设默认值 None 或具体值
    description: Optional[str] = Field(None, max_length=500, description="产品描述")
    # 数值约束：大于0，小于等于10000
    price: float = Field(..., gt=0, le=10000, description="产品价格")
    # 正则约束：字母数字下划线开头，后续可包含字母数字下划线
    category: str = Field(..., regex=r"^[a-zA-Z_][a-zA-Z0-9_]*$", description="产品分类")
    # 列表约束：最多10个标签
    tags: List[str] = Field(default=[], max_items=10, description="产品标签")
    # 使用工厂函数生成默认时间，避免固定值
    created_at: datetime = Field(default_factory=datetime.utcnow)
    is_active: bool = True

@app.post("/products/")
def create_product(product: ProductCreate):
    """
    创建产品
    自动完成JSON解析、类型转换、数据验证
    """
    # Pydantic 2.x 推荐使用 model_dump() 替代 dict()
    return product.model_dump()

自定义业务验证

当内置约束不够用时，可以用 @validator 处理单个字段，或用 @root_validator 进行多字段联动，实现复杂的业务逻辑。

from pydantic import validator, root_validator

class OrderItem(BaseModel):
    product_id: int
    quantity: int = Field(..., gt=0, le=1000)
    unit_price: float = Field(..., gt=0)

class OrderCreate(BaseModel):
    customer_id: int
    items: List[OrderItem] = Field(..., min_items=1)
    order_date: datetime
    delivery_date: datetime
    discount_code: Optional[str] = None

    @validator('discount_code')
    def validate_discount(cls, v):
        """单字段验证：折扣码格式检查"""
        if v and not re.match(r"^[A-Z0-9]{6,10}$", v):
            raise ValueError('折扣码格式无效（需要6-10位大写字母和数字）')
        return v.upper() if v else v

    @root_validator
    def validate_order_logic(cls, values):
        """多字段联动验证：业务规则"""
        order_date = values.get('order_date')
        delivery_date = values.get('delivery_date')
        discount_code = values.get('discount_code')

        # 配送时间必须严格晚于下单时间
        if order_date and delivery_date and delivery_date <= order_date:
            raise ValueError('配送时间必须晚于下单时间')
        # 订单总金额未满100元时，不能使用折扣码
        if discount_code and values.get('items'):
            total = sum(i.quantity * i.unit_price for i in values['items'])
            if total < 100:
                raise ValueError('小额订单（<100元）不能使用折扣码')
        return values

响应模型定义

使用 response_model 可以严格管控返回给客户端的字段，既避免泄露密码等敏感信息，也让响应结构更加统一。

from pydantic import EmailStr     # Pydantic 2.x 内置的邮箱验证
from fastapi import Path
from typing import TypeVar, Generic, Optional
from datetime import datetime, timezone

# 去除敏感字段后的用户响应模型
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    created_at: datetime
    is_active: bool
    # 注意：password字段已被排除在外，不会出现在响应中

# 泛型API响应模型，统一整体的返回格式
T = TypeVar('T')
class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "操作成功"
    data: Optional[T] = None
    timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

# 模拟用户数据库
fake_db = {
    1: {
        "id": 1,
        "username": "johndoe",
        "email": "john@example.com",
        "password": "secret",
        "created_at": datetime.now(timezone.utc),
        "is_active": True
    }
}

@app.get("/users/{user_id}", response_model=ApiResponse[UserResponse])
def get_user(user_id: int = Path(..., gt=0)):
    """获取用户，返回统一格式的响应"""
    user_data = fake_db.get(user_id)
    if not user_data:
        return ApiResponse[UserResponse](success=False, message="用户不存在", data=None)
    return ApiResponse[UserResponse](data=user_data)

文件上传与表单处理

基础文件上传

FastAPI 的 UploadFile 专为大文件设计，采用流式读取，内存占用极低。你可以轻松处理单个或多个文件。

from fastapi import File, UploadFile
from typing import List
import aiofiles
from pathlib import Path

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/uploadfile/")
async def upload_file(file: UploadFile = File(...)):
    """上传单个文件并异步保存"""
    # 生成唯一文件名，防止覆盖
    unique_name = f"{file.filename.split('.')[0]}_{hash(file.filename)}.{file.filename.split('.')[-1]}"
    save_path = UPLOAD_DIR / unique_name

    # 分块流式保存，每次读取 1MB
    async with aiofiles.open(save_path, 'wb') as f:
        while content := await file.read(1024 * 1024):
            await f.write(content)

    return {"filename": unique_name, "size": save_path.stat().st_size}

@app.post("/uploadfiles/")
async def upload_files(files: List[UploadFile] = File(...)):
    """一次上传多个文件"""
    results = []
    for file in files:
        size = len(await file.read())
        results.append({
            "filename": file.filename,
            "content_type": file.content_type,
            "size": size
        })
    return results

表单数据处理

处理普通的 HTML 表单，或 multipart/form-data 格式的非 JSON 数据，只需在参数上添加 Form(...)。

from fastapi import Form

@app.post("/login/")
async def login(username: str = Form(...), password: str = Form(...)):
    """基础表单登录"""
    return {"username": username, "message": "登录请求接收成功"}

嵌套模型与复杂数据

真实业务中很少遇到扁平的数据结构。嵌套模型能帮你清晰地描述订单、客户、地址等复杂关系。

class Address(BaseModel):
    street: str
    city: str
    country: str
    postal_code: str

class CustomerResponse(BaseModel):
    id: int
    name: str
    email: EmailStr
    address: Optional[Address] = None

class OrderItemResponse(BaseModel):
    product_name: str
    quantity: int
    total_price: float

class OrderResponse(BaseModel):
    order_id: str
    customer: CustomerResponse
    items: List[OrderItemResponse]
    total_amount: float
    status: str

# 模拟一个包含嵌套数据的完整订单
sample_order = OrderResponse(
    order_id="ORD2024010001",
    customer=CustomerResponse(
        id=1,
        name="John Doe",
        email="john@example.com",
        address=Address(street="123 Main St", city="New York", country="USA", postal_code="10001")
    ),
    items=[
        OrderItemResponse(product_name="Laptop", quantity=1, total_price=999.99),
        OrderItemResponse(product_name="Mouse", quantity=2, total_price=59.98)
    ],
    total_amount=1059.97,
    status="processing"
)

错误处理与性能优化

请求体验证错误处理

通过自定义 RequestValidationError 的exception-handling器，可以把枯燥的默认错误转换成结构清晰、对调用方友好的格式。

from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    """统一结构化验证错误响应"""
    errors = []
    for err in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in err["loc"][1:]),  # 去除 body 前缀
            "message": err["msg"],
            "type": err["type"]
        })
    return JSONResponse(
        status_code=422,
        content=ApiResponse[None](
            success=False,
            message="请求数据验证失败",
            data={"errors": errors}
        ).model_dump()
    )

基础性能优化建议

  • 善用 async：文件读写、数据库查询、外部 API 调用等 IO 密集型操作，全部使用 async/await 避免阻塞。
  • 精简验证：不要在请求模型中滥用极复杂的正则表达式，能用内置类型约束解决的问题优先使用内置方法。
  • 大文件分块上传：通过 UploadFile.read(size) 按块处理，避免一次性将所有内容读入内存。
  • 控制响应体积：恰当使用 response_model_exclude_unset 或 response_model_exclude_defaults，只返回真正有意义的字段。

实际应用案例

下面是一个完整的 创建订单 API，融合了前面介绍的各种技巧：请求模型验证、响应模型控制、后台任务处理等。

from fastapi import BackgroundTasks
import uuid
import asyncio

# 模拟数据库操作
async def save_order_to_db(order_data: dict):
    print(f"保存订单到数据库: {order_data}")
    await asyncio.sleep(0.1)

# 模拟邮件通知
async def send_order_confirmation(order_id: str, email: str):
    print(f"发送订单确认邮件到 {email}，订单号 {order_id}")
    await asyncio.sleep(0.2)

@app.post("/orders/", response_model=ApiResponse[OrderResponse])
async def create_order_api(
    order: OrderCreate,
    background_tasks: BackgroundTasks
):
    """完整的创建订单 API"""
    # 生成唯一的订单号
    order_id = f"ORD-{uuid.uuid4().hex[:8].upper()}"
    # 计算订单总金额
    total_amount = sum(item.quantity * item.unit_price for item in order.items)
    # 组装响应数据（示例中写死部分客户信息，生产环境应查询数据库）
    order_response = OrderResponse(
        order_id=order_id,
        customer=CustomerResponse(
            id=order.customer_id,
            name="John Doe",
            email="john@example.com",
            address=Address(street="123 Main St", city="New York", country="USA", postal_code="10001")
        ),
        items=[
            OrderItemResponse(
                product_name=f"Product {item.product_id}",
                quantity=item.quantity,
                total_price=item.quantity * item.unit_price
            )
            for item in order.items
        ],
        total_amount=total_amount,
        status="processing"
    )
    # 将耗时操作放入后台任务，避免阻塞接口返回
    background_tasks.add_task(save_order_to_db, order.model_dump())
    background_tasks.add_task(send_order_confirmation, order_id, "john@example.com")
    return ApiResponse[OrderResponse](data=order_response, message="订单创建成功")

相关教程

:::tip request-body-handling最佳实践

  1. 明确区分「请求模型」和「响应模型」，避免无意中返回敏感信息
  2. 优先使用 Pydantic 2.x 的内置验证器（如 EmailStr、PaymentCardNumber），减少自定义正则
  3. 多字段联动验证交给 @root_validator，保持业务逻辑的完整性
  4. 大文件上传务必使用 UploadFile.read(size) 分块读取，防止内存溢出 :::

总结

FastAPI 的请求体与响应模型是它最亮眼的能力之一。借助 Pydantic 2.x 的强劲动力，你只需少量代码就能获得安全、规范且高效的数据传输体验。熟练运用这些技巧，就能快速构建出企业级的 Web API。