FastAPI请求体处理与响应模型

📂 所属阶段：第一阶段 — 快速筑基（基础篇）
🔗 相关章节：路径参数与查询参数 · 响应处理与状态码 · Pydantic综合指南

目录

请求体处理概述

请求体是Web API接收客户端数据的核心载体，FastAPI结合Pydantic 2.x提供了一套「自动验证+自动文档+类型转换」三位一体的解决方案，完全避免了传统框架中手动解析JSON/表单、手写验证代码的繁琐。

核心处理流程

  1. Pydantic模型定义数据契约：明确字段类型、必填性、约束规则
  2. FastAPI自动解析与验证：从HTTP请求中提取数据，转换为Python类型并校验
  3. 错误友好提示：校验失败时返回结构化的错误信息，方便客户端定位问题
  4. OpenAPI/Swagger文档自动生成：模型约束直接映射为API文档的示例与参数说明

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解析、类型转换、数据验证
    """
    return product.model_dump()  # Pydantic 2.x推荐使用model_dump替代dict()

自定义业务验证

内置约束不够用时，使用@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('配送时间必须晚于下单时间')
        # 小额订单禁用折扣码
        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参数可以严格控制返回给客户端的字段，避免泄露敏感信息（如密码、内部ID），同时统一响应格式：

from pydantic import EmailStr  # 更简洁的内置邮箱验证（Pydantic 2.x）
from fastapi import Path

# 敏感信息排除的用户响应模型
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    created_at: datetime
    is_active: bool
    # password字段不会出现在响应中

# 泛型API响应模型（统一响应结构）
from typing import Generic, TypeVar, Optional

T = TypeVar('T')
class ApiResponse(BaseModel, Generic[T]):
    success: bool = True
    message: str = "操作成功"
    data: Optional[T] = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

# 模拟用户数据库
fake_db = {1: {"id": 1, "username": "johndoe", "email": "john@example.com", "password": "secret", "created_at": datetime.utcnow(), "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

    # 异步流式保存（适合大文件）
    async with aiofiles.open(save_path, 'wb') as f:
        while content := await file.read(1024*1024):  # 每次读取1MB
            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数据：

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的响应格式，提高API的可读性：

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/await处理IO密集型操作（如文件读写、数据库查询、外部API调用）
  • 避免在请求模型中使用不必要的复杂验证（如非常长的正则表达式）
  • 大文件上传时使用UploadFile.read(size)分块读取
  • 合理使用response_model_exclude_unset/response_model_exclude_defaults减少响应体积

实际应用案例

结合前面的内容，实现一个创建订单的完整API：

from fastapi import BackgroundTasks
import uuid

# 模拟数据库操作
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(i.quantity * i.unit_price for i 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 {i.product_id}", quantity=i.quantity, total_price=i.quantity*i.unit_price) for i 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="订单创建成功")

相关教程

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

总结

FastAPI的请求体与响应模型是其核心竞争力之一，结合Pydantic 2.x的强大功能，开发者可以用最少的代码实现「安全、规范、高效」的数据传输。掌握这些技巧，就能快速构建企业级的Web API。