1. 请求体参数基础

1.1 什么是请求体参数

请求体参数是通过 HTTP 请求体（request body）传递给服务器的数据，通常用于发送复杂的数据结构，如 JSON 对象、表单数据等。与查询参数和路径参数不同，请求体参数可以包含大量数据，并且可以使用不同的编码格式。

1.2 请求体参数的基本语法

在 FastAPI 中，定义请求体参数非常简单，只需在路径操作函数的参数中使用 oydantic模型即可：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(item: Item):
    return {"item": item}

FastAPI 会自动将请求体中的 JSON 数据解析为 Pydantic 模型对象。

1.3 参数的对比

特性	路径参数	查询参数	请求体参数
位置	URL 路径中	URL 末尾的查询字符串	HTTP 请求体中
大小限制	受 URL 长度限制	受 URL 长度限制	通常无限制（可配置）
数据类型	简单类型	简单类型	复杂类型（JSON、Form Data等）
用途	标识资源	过滤、分页、排序等	创建、更新资源
安全性	可见于 URL	可见于 URL	不可见于 URL

1.4 参数混合使用

在 FastAPI 中，可以在同一个端点中混合使用请求体参数、查询参数和路径参数：

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.put("/items/{item_id}")
def update_item(
    item_id: int,  # 路径参数
    item: Item,  # 请求体参数
    q: str = Query(None, description="查询参数")  # 查询参数
):
    """更新项目
    
    - **item_id**: 项目 ID
    - **item**: 项目信息
    - **q**: 查询参数
    """
    return {
        "item_id": item_id,
        "item": item,
        "q": q
    }

FastAPI 会自动识别不同类型的参数，并从相应的位置提取数据。

1.5 请求体编码格式

FastAPI 支持多种请求体编码格式：

1. JSON：默认格式，适用于大多数 API 场景
2. Form Data：适用于表单提交
3. Multipart Form：适用于文件上传
4. XML：需要额外配置

1.6 简单示例

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    username: str
    email: str
    full_name: str = None
    disabled: bool = False

@app.post("/users/")
def create_user(user: User):
    """创建新用户
    
    - **username**: 用户名
    - **email**: 邮箱地址
    - **full_name**: 全名（可选）
    - **disabled**: 是否禁用（可选，默认为 False）
    """
    return {"user": user}

2. 类型系统与数据验证

2.1 基础数据类型

FastAPI 支持多种基础数据类型作为请求体参数：

• str: 字符串类型
• int: 整数类型
• float: 浮点数类型
• bool: 布尔类型
• None: 空值

2.2 高级类型

• 列表类型: List[str] 或 List[int]
• 字典类型: Dict[str, Any]
• 枚举类型: 继承自 Enum 的类
• 嵌套结构: 使用 Pydantic 模型
• 联合类型: Union[str, int]
• 可选类型: Optional[str]

2.3 Pydantic 模型的使用

Pydantic 是 FastAPI 的核心依赖之一，用于数据验证和设置管理。使用 Pydantic 模型可以定义复杂的请求体结构：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: List[str] = []

@app.post("/items/")
def create_item(item: Item):
    return {"item": item}

2.4 Body 参数详解

FastAPI 提供了 Body 函数用于更细粒度地控制请求体参数：

from fastapi import FastAPI, Body
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
def create_item(
    item: Item = Body(..., embed=True, description="项目信息")
):
    return {"item": item}

Body 函数的主要参数：

• default: 默认值
• embed: 是否将参数嵌入到请求体中
• media_type: 媒体类型
• description: 参数描述
• example: 示例值
• alias: 参数别名

2.5 字段验证规则

Pydantic 提供了丰富的字段验证规则：

from fastapi import FastAPI
from pydantic import BaseModel, Field

app = FastAPI()

class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, description="项目名称")
    price: float = Field(..., ge=0, description="项目价格")
    tax: float = Field(None, ge=0, description="税率")

@app.post("/items/")
def create_item(item: Item):
    return {"item": item}

常用的验证规则：

• min_length/max_length: 字符串最小/最大长度
• ge/gt/le/lt: 数值大于等于/大于/小于等于/小于
• regex: 正则表达式验证
• default: 默认值
• description: 字段描述
• example: 示例值

2.6 类型转换与自动解析

FastAPI 会自动处理类型转换，例如：

• 字符串 "123" 会被转换为整数 123
• 字符串 "true" 会被转换为布尔值 True
• JSON 对象会被转换为 Pydantic 模型对象

3. 复杂请求体结构

3.1 嵌套模型

Pydantic 模型可以嵌套使用，创建复杂的请求体结构：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI()

class Address(BaseModel):
    street: str
    city: str
    zipcode: str

class User(BaseModel):
    username: str
    email: str
    address: Address
    phone_numbers: List[str] = []

@app.post("/users/")
def create_user(user: User):
    return {"user": user}

3.2 列表类型请求体

请求体可以是列表类型：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/batch")
def create_items(items: List[Item]):
    return {"items": items, "count": len(items)}

3.3 字典类型请求体

请求体可以是字典类型：

from fastapi import FastAPI
from typing import Dict, Any

app = FastAPI()

@app.post("/data/")
def create_data(data: Dict[str, Any]):
    return {"data": data}

3.4 混合类型请求体

请求体可以是混合类型：

from fastapi import FastAPI, Body
from pydantic import BaseModel
from typing import List, Dict, Any, Optional

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/mixed/")
def create_mixed(
    items: List[Item] = Body(..., description="项目列表"),
    metadata: Dict[str, Any] = Body(..., description="元数据"),
    tags: List[str] = Body(default_factory=list, description="标签列表")
):
    return {"items": items, "metadata": metadata, "tags": tags}

3.5 可选字段与默认值

可以为字段设置默认值，使其成为可选字段：

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

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    description: Optional[str] = None
    tags: List[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.utcnow)

@app.post("/items/")
def create_item(item: Item):
    return {"item": item}

4. 依赖注入与请求体

4.1 依赖注入与请求体的结合

FastAPI 的依赖注入系统可以与请求体无缝集成：

from fastapi import FastAPI, Depends, Body
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

def item_validator(item: Item = Body(...)):
    # 验证项目
    if item.price <= 0:
        raise ValueError("价格必须为正数")
    return item

@app.post("/items/")
def create_item(item: Item = Depends(item_validator)):
    return {"item": item}

4.2 共享依赖的实现

可以将通用的请求体处理逻辑提取为共享依赖：

from fastapi import FastAPI, Depends, Body
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

class Pagination(BaseModel):
    skip: int = 0
    limit: int = 10

class Sorting(BaseModel):
    sort_by: str = "id"
    sort_order: str = "asc"

class Filter(BaseModel):
    name: Optional[str] = None
    min_price: Optional[float] = None
    max_price: Optional[float] = None

class QueryParams(BaseModel):
    pagination: Pagination
    sorting: Sorting
    filter: Filter

def get_query_params(
    pagination: Pagination = Body(default_factory=Pagination),
    sorting: Sorting = Body(default_factory=Sorting),
    filter: Filter = Body(default_factory=Filter)
):
    return QueryParams(pagination=pagination, sorting=sorting, filter=filter)

@app.post("/items/")
def get_items(params: QueryParams = Depends(get_query_params)):
    return {"params": params}

4.3 动态请求体处理

对于不确定结构的请求体，可以使用 Any 类型：

from fastapi import FastAPI, Body
from typing import Any

app = FastAPI()

@app.post("/dynamic/")
def create_dynamic(data: Any = Body(...)):
    return {"data": data, "type": type(data).__name__}

4.4 复杂业务逻辑的封装

依赖注入还可以用于封装复杂的业务逻辑：

from fastapi import FastAPI, Depends, Body
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

class Order(BaseModel):
    items: List[Item]
    user_id: int

class OrderService:
    def __init__(self, order: Order = Body(...)):
        self.order = order
    
    def process_order(self):
        # 计算总价
        total = sum(item.price for item in self.order.items)
        # 验证用户
        # 处理支付
        # 生成订单
        return {
            "order_id": 123,
            "user_id": self.order.user_id,
            "items": self.order.items,
            "total": total
        }

@app.post("/orders/")
def create_order(order_service: OrderService = Depends()):
    return order_service.process_order()

5. 性能优化策略

5.1 请求体解析性能分析

FastAPI 的请求体解析性能主要受以下因素影响：

• 数据大小：请求体越大，解析时间越长
• 数据复杂度：嵌套结构越复杂，解析时间越长
• 验证规则：验证规则越多，解析时间越长
• 编码格式：不同编码格式的解析性能不同

5.2 缓存应用

对于频繁使用的请求体结构，可以使用缓存来提高性能：

from functools import lru_cache
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@lru_cache(maxsize=1000)
def process_item(item: Item):
    # 处理项目
    return {"processed": True, "item": item}

@app.post("/items/")
def create_item(item: Item):
    return process_item(item)

5.3 大数据量请求体处理

处理大数据量请求体的策略：

1. 流式读取：使用流式读取，避免一次性加载整个请求体到内存
2. 分块处理：将大请求体分成多个小块进行处理
3. 异步处理：使用异步代码处理请求体
4. 数据库优化：使用批量插入等优化数据库操作
5. 缓存：使用缓存减少重复处理

5.4 请求体大小限制与处理

设置请求体大小限制：

from fastapi import FastAPI, Request
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()

# 启用 GZIP 压缩
app.add_middleware(GZipMiddleware, minimum_size=1000)

# 自定义中间件限制请求体大小
class RequestSizeLimitMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        if "content-length" in request.headers:
            content_length = int(request.headers["content-length"])
            if content_length > 10 * 1024 * 1024:  # 10MB 限制
                return JSONResponse(
                    status_code=413,
                    content={"detail": "Request body too large"}
                )
        response = await call_next(request)
        return response

app.add_middleware(RequestSizeLimitMiddleware)

5.5 批量操作优化

对于批量操作，使用列表类型的请求体可以提高性能：

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items/batch")
def create_items(items: List[Item]):
    # 批量处理项目
    return {
        "count": len(items),
        "items": items
    }

6. 源码深度分析

6.1 处理请求体的内部机制

1. 请求接收：FastAPI 接收到 HTTP 请求
2. 路由匹配：根据 URL 路径匹配对应的路径操作函数
3. 内容类型解析：根据 Content-Type 头确定请求体编码格式
4. 请求体读取：读取并解析请求体数据
5. 类型转换：将请求体数据转换为指定的类型
6. 验证：使用 Pydantic 进行数据验证
7. 依赖注入：处理依赖项，包括请求体依赖
8. 执行函数：调用路径操作函数并传递解析后的参数
9. 响应生成：将函数返回值转换为 HTTP 响应

6.2 执行流程图

6.3 关键类与方法分析

1. FastAPI 类：继承自 Starlette，添加了 FastAPI 特有的功能
2. BaseModel 类：Pydantic 模型的基类，用于数据验证
3. Body 函数：用于定义请求体参数
4. Request 类：表示 HTTP 请求，包含请求体
5. Form 函数：用于定义表单参数
6. File 函数：用于定义文件上传参数
7. solve_dependencies 函数：处理依赖项

6.4 与 Starlette 的交互

FastAPI 基于 Starlette 构建，利用 Starlette 的请求处理功能，同时添加了：

• 类型提示支持
• Pydantic 数据验证
• 自动 API 文档生成
• 依赖注入系统

6.5 源码优化建议

6.5.1 缓存优化

增加请求体解析结果的缓存，特别是对于频繁使用的请求体结构：

from functools import lru_cache
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@lru_cache(maxsize=1000)
def process_item(item: Item):
    # 处理项目
    return {"processed": True, "item": item}

@app.post("/items/")
def create_item(item: Item):
    return process_item(item)

6.5.2 并行处理

对于复杂的请求体解析，考虑使用并行处理提高性能：

import asyncio
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

async def process_single_item(item: Item):
    # 处理单个项目
    await asyncio.sleep(0.1)  # 模拟耗时操作
    return {"processed": True, "item": item}

@app.post("/items/batch")
async def create_items(items: List[Item]):
    # 并行处理多个项目
    tasks = [process_single_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return {"results": results, "count": len(results)}

6.5.3 内存优化

减少请求体解析过程中的内存使用，特别是对于大型请求体：

from fastapi import FastAPI, Request
import ujson

app = FastAPI()

@app.post("/large-data/")
async def process_large_data(request: Request):
    # 流式读取请求体，避免一次性加载到内存
    body = await request.body()
    # 使用 ujson 进行高效解析
    data = ujson.loads(body)
    # 处理数据
    return {"status": "processed", "size": len(body)}

6.5.4 预编译

预编译验证规则，减少运行时开销：

import re
from fastapi import FastAPI
from pydantic import BaseModel, field_validator

app = FastAPI()

# 预编译正则表达式
EMAIL_REGEX = re.compile(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$')

class User(BaseModel):
    email: str
    username: str
    
    @field_validator('email')
    @classmethod
    def email_must_be_valid(cls, v):
        if not EMAIL_REGEX.match(v):
            raise ValueError('邮箱格式无效')
        return v

@app.post("/users/")
def create_user(user: User):
    return {"user": user}

6.5.5 延迟解析

对于复杂的请求体结构，使用延迟解析策略，只在需要时才解析：

from fastapi import FastAPI, Body
from typing import Any, Optional

app = FastAPI()

@app.post("/dynamic/")
async def create_dynamic(data: Any = Body(...)):
    # 延迟解析：只在需要时才访问具体字段
    if "type" in data:
        if data["type"] == "user":
            # 处理用户数据
            return {"type": "user", "name": data.get("name")}
        elif data["type"] == "product":
            # 处理产品数据
            return {"type": "product", "price": data.get("price")}
    return {"type": "unknown", "data": data}

7. 常见问题与解决方案

7.1 类型转换错误

问题：请求体类型转换失败，例如将字符串转换为整数时出错。

解决方案：

• 使用 Pydantic 模型进行类型验证
• 提供合理的默认值
• 使用 Optional 类型处理可选字段
• 自定义验证器处理特殊情况

7.2 验证失败处理

问题：请求体验证失败，返回 422 错误。

解决方案：

• 提供明确的字段描述和示例
• 使用 Pydantic 模型进行更复杂的验证
• 自定义错误处理逻辑
• 前端添加表单验证，减少无效请求

7.3 复杂嵌套结构的问题

问题：复杂的嵌套请求体结构难以处理。

解决方案：

• 使用 Pydantic 模型定义复杂结构
• 合理设计模型层次，避免过深的嵌套
• 使用依赖注入处理复杂逻辑
• 考虑使用 GraphQL 等更灵活的 API 设计

7.4 性能瓶颈分析

问题：请求体解析成为性能瓶颈。

解决方案：

• 使用缓存机制
• 减少请求体的大小和复杂度
• 优化依赖注入链
• 使用异步代码处理请求体
• 考虑使用更高效的序列化库

7.5 安全风险与防范

问题：请求体可能导致安全风险，例如注入攻击。

解决方案：

• 对请求体进行严格验证
• 使用参数化查询，避免 SQL 注入
• 限制请求体的大小和数量
• 使用 HTTPS 保护传输中的数据
• 实现请求速率限制，防止 DoS 攻击

8. 生产最佳实践与案例分析

8.1 安全设置

• 避免注入攻击：使用参数化查询，避免直接拼接 SQL
• 输入验证：对所有请求体参数进行严格验证
• 请求体大小限制：限制请求体的大小，防止 DoS 攻击
• HTTPS：使用 HTTPS 保护传输中的数据
• CORS 配置：合理配置 CORS，避免跨域攻击

8.2 文档自动生成

• 使用描述：为请求体参数添加详细的 description
• 提供示例：使用 example 参数提供示例值
• 分组参数：使用 Pydantic 模型对相关参数进行分组
• API 文档：利用 FastAPI 自动生成的 API 文档

8.3 测试策略

• 单元测试：测试请求体解析和验证逻辑
• 集成测试：测试完整的 API 端点
• 边界测试：测试参数的边界值
• 性能测试：测试请求体解析的性能
• 安全测试：测试请求体相关的安全漏洞

8.4 实际项目案例分析

8.4.1 用户注册与登录

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, field_validator
from typing import Optional

app = FastAPI()

class UserRegister(BaseModel):
    username: str
    email: str
    password: str
    confirm_password: str
    
    @field_validator('email')
    @classmethod
    def email_must_be_valid(cls, v):
        if '@' not in v:
            raise ValueError('邮箱格式无效')
        return v
    
    @field_validator('confirm_password')
    @classmethod
    def passwords_must_match(cls, v, info):
        if 'password' in info.data and v != info.data['password']:
            raise ValueError('两次输入的密码不一致')
        return v

class UserLogin(BaseModel):
    email: str
    password: str

@app.post("/register/")
def register(user: UserRegister):
    # 检查用户是否已存在
    # 密码加密
    # 创建用户
    return {"message": "注册成功", "username": user.username}

@app.post("/login/")
def login(user: UserLogin):
    # 验证用户
    # 生成 token
    return {"message": "登录成功", "token": "example_token"}

8.4.2 数据创建与更新

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

class ItemBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: Optional[str] = None
    price: float = Field(..., ge=0)

class ItemCreate(ItemBase):
    pass

class ItemUpdate(BaseModel):
    name: Optional[str] = Field(None, min_length=1, max_length=100)
    description: Optional[str] = None
    price: Optional[float] = Field(None, ge=0)

# 模拟数据库
items = {}

@app.post("/items/")
def create_item(item: ItemCreate):
    item_id = len(items) + 1
    items[item_id] = item
    return {"id": item_id, "item": item}

@app.put("/items/{item_id}")
def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="项目不存在")
    
    # 更新项目
    existing_item = items[item_id]
    update_data = item.dict(exclude_unset=True)
    updated_item = existing_item.copy(update=update_data)
    items[item_id] = updated_item
    
    return {"id": item_id, "item": updated_item}

8.4.3 批量操作实现

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

class BatchResponse(BaseModel):
    success: bool
    count: int
    items: List[Item]

@app.post("/items/batch", response_model=BatchResponse)
def create_items(items: List[Item]):
    # 批量处理项目
    # 这里可以添加数据库批量插入逻辑
    
    return BatchResponse(
        success=True,
        count=len(items),
        items=items
    )

8.4.4 文件上传与处理

from fastapi import FastAPI, File, UploadFile, Form
from typing import List, Optional
import shutil
import os
import uuid

app = FastAPI()

# 确保上传目录存在
os.makedirs("uploads", exist_ok=True)

@app.post("/upload/")
async def upload_files(
    files: List[UploadFile] = File(..., description="上传文件"),
    description: Optional[str] = Form(None, description="文件描述")
):
    uploaded_files = []
    
    for file in files:
        # 生成唯一文件名
        file_id = str(uuid.uuid4())
        file_extension = os.path.splitext(file.filename)[1]
        file_path = f"uploads/{file_id}{file_extension}"
        
        # 保存文件
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)
        
        uploaded_files.append({
            "filename": file.filename,
            "file_id": file_id,
            "path": file_path,
            "content_type": file.content_type
        })
    
    return {
        "description": description,
        "uploaded_files": uploaded_files,
        "count": len(uploaded_files)
    }

8.4.5 复杂业务场景应用

from fastapi import FastAPI, Depends, Body
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI()

class Product(BaseModel):
    id: int
    name: str
    price: float
    stock: int

class OrderItem(BaseModel):
    product_id: int
    quantity: int

class Order(BaseModel):
    user_id: int
    items: List[OrderItem]
    shipping_address: str
    payment_method: str

class OrderService:
    def __init__(self, order: Order = Body(...)):
        self.order = order
        self.products = {
            1: Product(id=1, name="Product 1", price=10.0, stock=100),
            2: Product(id=2, name="Product 2", price=20.0, stock=50),
            3: Product(id=3, name="Product 3", price=30.0, stock=25)
        }
    
    def process_order(self):
        # 验证库存
        total = 0
        for item in self.order.items:
            if item.product_id not in self.products:
                raise ValueError(f"产品 {item.product_id} 不存在")
            product = self.products[item.product_id]
            if product.stock < item.quantity:
                raise ValueError(f"产品 {product.name} 库存不足")
            total += product.price * item.quantity
        
        # 扣减库存
        for item in self.order.items:
            product = self.products[item.product_id]
            product.stock -= item.quantity
        
        # 生成订单
        order_id = int(datetime.utcnow().timestamp() * 1000)
        
        return {
            "order_id": order_id,
            "user_id": self.order.user_id,
            "items": self.order.items,
            "shipping_address": self.order.shipping_address,
            "payment_method": self.order.payment_method,
            "total": total,
            "status": "pending"
        }

@app.post("/orders/")
def create_order(order_service: OrderService = Depends()):
    return order_service.process_order()

8.4.6 用户配置更新案例

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any

app = FastAPI()

class NotificationSettings(BaseModel):
    email: bool = True
    sms: bool = False
    push: bool = True

class UserPreferences(BaseModel):
    theme: str = Field(default="light", pattern="^(light|dark)$")
    language: str = Field(default="en", min_length=2, max_length=2)
    notifications: NotificationSettings = Field(default_factory=NotificationSettings)
    preferences: Dict[str, Any] = Field(default_factory=dict)

class UserProfile(BaseModel):
    full_name: Optional[str] = None
    bio: Optional[str] = None
    avatar: Optional[str] = None
    preferences: UserPreferences = Field(default_factory=UserPreferences)

# 模拟数据库
users = {
    1: UserProfile(
        full_name="John Doe",
        bio="Software developer",
        avatar="https://example.com/avatar.jpg",
        preferences=UserPreferences(
            theme="dark",
            language="zh",
            notifications=NotificationSettings(
                email=True,
                sms=True,
                push=True
            ),
            preferences={"notifications_frequency": "daily"}
        )
    )
}

@app.get("/users/{user_id}/profile")
def get_user_profile(user_id: int):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="用户不存在")
    return users[user_id]

@app.put("/users/{user_id}/profile")
def update_user_profile(user_id: int, profile: UserProfile):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="用户不存在")
    users[user_id] = profile
    return {"message": "用户配置更新成功", "profile": profile}

@app.patch("/users/{user_id}/preferences")
def update_user_preferences(user_id: int, preferences: UserPreferences):
    if user_id not in users:
        raise HTTPException(status_code=404, detail="用户不存在")
    users[user_id].preferences = preferences
    return {"message": "用户偏好设置更新成功", "preferences": preferences}

这个案例展示了如何处理复杂的嵌套请求体结构，包括：

• 嵌套的 Pydantic 模型
• 字段验证规则
• 默认值和默认工厂
• 部分更新（PATCH 请求）
• 错误处理