FastAPI请求体与Pydantic模型

在FastAPI中，请求体（RequestBody）用于接收客户端发送的结构化数据（如JSON），通常与Pydantic模型结合实现数据验证和自动文档生成。

请求体基础用法

定义Pydantic模型

from pydantic import BaseModel

class ItemCreate(BaseModel):
    name: str
    price: float
    description: str | None = None  # 可选字段
    tags: list[str] = []  # 默认空列表

在路由中使用模型

from fastapi import FastAPI

app = FastAPI()

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

请求示例：

{
    "name": "Laptop",
    "price": 5999.99,
    "description": "高性能笔记本",
    "tags": ["electronics","sale"]
}

• 自动验证：缺失必填字段或类型错误会触发HTTP422错误
• Swagger文档：自动生成交互式请求示例

高级请求体配置

多请求体参数

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,  # 路径参数
    item: ItemCreate,  # 请求体模型
    q: str | None = None  # 查询参数
):
    return {"item_id": item_id, **item.dict(), "q": q}

单请求体多模型

from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    email: EmailStr

class UserCreate(UserBase):
    password: str

class UserUpdate(UserBase):
    name: str | None
    is_active: bool = True

文件上传与表单数据

文件上传

from fastapi import UploadFile

@app.post("/upload/")
async def upload_file(
    file: UploadFile,
    description: str = Form(...)
):
    return {
        "filename": file.filename,
        "size": len(await file.read()),
        "description": description
    }

混合表单与JSON

from fastapi import Form
from pydantic import BaseModel

class ItemForm(BaseModel):
    name: str
    price: float = Form(..., gt=0)

@app.post("/form-items/")
async def form_item(item: ItemForm = Depends()):
    return item

Pydantic模型

Pydantic模型是FastAPI生态的核心组件，用于数据验证、序列化和文档生成。通过合理运用Pydantic模型，可以实现：

• 数据验证：确保输入数据符合业务规则
• 自动文档：生成准确的API文档
• 类型安全：减少运行时类型错误
• 序列化控制：灵活处理数据输入输出格式

结合FastAPI的依赖注入系统，能够构建出健壮且易维护的API服务。

基础模型定义

必填字段与可选字段

from pydantic import BaseModel
from typing import Optional

class User(BaseModel):
    username: str  # 必填字段
    email: Optional[str] = None  # 可选字段
    age: int = 18  # 带默认值的必填字段

字段类型增强

from pydantic import HttpUrl, PaymentCardNumber

class Profile(BaseModel):
    website: HttpUrl  # 自动验证URL格式
    phone: str = Field(regex=r"^1[3-9]\d{9}$")  # 正则验证手机号
    credit_card: PaymentCardNumber  # 专用类型验证

严格类型校验

使用 strict 参数禁止类型自动转换（如字符串”123″转整数123）。

示例：严格模式

from pydantic import BaseModel, StrictInt

class Data(BaseModel):
    number: StrictInt  # 必须严格为整数类型

验证机制详解

内置验证器

通过继承 BaseModel 定义数据模型，字段类型和默认值决定了验证规则。Pydantic支持丰富的字段类型和内置校验。

常见字段类型

from pydantic import Field

class Product(BaseModel):
    name: str = Field(min_length=2, max_length=50)
    price: float = Field(gt=0, le=1000000)
    stock: int = Field(ge=0)

自定义验证器

使用 @validator 装饰器定义字段级别的自定义验证逻辑。

示例：验证密码强度

from pydantic import BaseModel, validator

class User(BaseModel):
    password: str

    @validator("password")
    def validate_password(cls, value):
        if len(value) < 8:
            raise ValueError("密码长度必须≥8")
        if not any(c.isupper() for c in value):
            raise ValueError("密码必须包含大写字母")
        return value

说明：

• @validator(“password”)表示验证 password 字段。
• 若验证失败，抛出ValueError，错误信息会返回给调用方。

根验证器

对整个模型的数据进行验证，可访问所有字段的值。

示例：验证字段一致性

from pydantic import BaseModel, root_validator

class UserRegistration(BaseModel):
    password: str
    confirm_password: str

    @root_validator
    def check_passwords_match(cls, values):
        if values.get("password") != values.get("confirm_password"):
            raise ValueError("两次密码不一致")
        return values

模型继承体系

嵌套模型

模型可以嵌套其他模型，处理复杂数据结构。

示例：嵌套模型验证

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

class User(BaseModel):
    name: str
    address: Address  # 嵌套Address模型

验证逻辑：

• address字段必须是一个包含 city 和 street 的字典。

基础模型复用

class BaseUser(BaseModel):
    email: EmailStr

class UserCreate(BaseUser):
    password: str

class UserResponse(BaseUser):
    id: int
    is_active: bool

联合类型

from typing import Union

class Cat(BaseModel):
    type: str = "cat"
    meow: bool

class Dog(BaseModel):
    type: str = "dog"
    bark: str

@app.post("/animals/")
async def adopt_animal(animal: Union[Cat, Dog]):
    return animal

混合继承

class TimestampMixin(BaseModel):
    created_at: datetime
    updated_at: datetime

class Article(TimestampMixin):
    title: str
    content: str

高级模型配置

ORM模式配置

class UserORM(BaseModel):
    id: int
    name: str

    class Config:
        orm_mode = True  # 允许从ORM对象转换

# 使用示例
user_orm = UserORM.from_orm(database_user)

字段别名处理

class DataModel(BaseModel):
    internal_name: str = Field(..., alias="externalName")

    class Config:
        allow_population_by_field_name = True  # 允许通过原始名称或别名赋值

敏感数据处理

class SecureData(BaseModel):
    password: str

    class Config:
        validate_assignment = True  # 赋值时触发验证
        extra = "forbid"  # 禁止额外字段
        json_encoders = {
            SecretStr: lambda v: v.get_secret_value() if v else None
        }

模型配置（Config类）

自定义模型行为：

class User(BaseModel):
    name: str

    class Config:
        allow_mutation = False  # 禁止修改实例字段
        anystr_strip_whitespace = True  # 自动去除字符串两端空格

自定义错误信息

通过 Field 类的参数自定义错误信息。

示例：自定义错误提示

from pydantic import BaseModel, Field

class User(BaseModel):
    age: int = Field(..., gt=0, description="用户年龄", example=25,
    error_messages={
        "gt": "年龄必须大于0",
        "type": "年龄必须是整数"
    })

动态模型技巧

动态默认值

使用 default_factory 生成动态默认值（如当前时间）。

示例：动态默认时间戳

from datetime import datetime
from pydantic import BaseModel, Field

class Post(BaseModel):
    title: str
    created_at: datetime = Field(default_factory=datetime.now)

运行时生成模型

from pydantic import create_model

DynamicModel = create_model(
    "DynamicModel",
    name=(str, ...),
    value=(float, 0.0),
    __config__=dict(extra="ignore")
)

条件字段模型

from typing import Literal

class Payment(BaseModel):
    method: Literal["credit", "alipay"]
    card_number: str | None = None
    alipay_account: str | None = None

    @validator("card_number", always=True)
    def check_card(cls, v, values):
        if values.get("method") == "credit" and not v:
            raise ValueError("信用卡支付需要卡号")
        return v

性能优化方案

模型缓存

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_model(fields: tuple):
    return create_model(f"CachedModel", **{f: (str, ...) for f in fields})

避免重复验证

class OptimizedModel(BaseModel):
    data: str

    def __init__(self, **data) -> None:
        if 'pre_validated' in data: #跳过已验证数据
            super().__init__(**data)
        else:
            #执行完整验证流程
            super().__init__(**data)

常见问题解决

循环引用问题

# post.py
from typing import List
from .user import User

class Post(BaseModel):
    author: "User"
    comments: List["Comment"]

# user.py
from .post import Post

class User(BaseModel):
    posts: List[Post]

时区处理

from pydantic import validator
from datetime import datetime, timezone

class Event(BaseModel):
    timestamp: datetime

    @validator("timestamp")
    def ensure_utc(cls, v):
        if v.tzinfo != timezone.utc:
            return v.astimezone(timezone.utc)
        return v

大数据量处理

from pydantic import parse_obj_as

#直接解析原始数据避免对象实例化开销
data_list = [{"id": i, "name": f"Item {i}"} for i in range(100000)]
items = parse_obj_as(list[Item], data_list)

最佳实践总结

分层设计模型

• BaseSchema: 基础字段定义
• CreateSchema: 创建专用字段
• UpdateSchema: 更新专用字段
• ResponseSchema: 响应数据结构

验证逻辑分离

class BusinessLogic:
    @staticmethod
    def validate_order(order: OrderCreate):
        #复杂业务规则验证
        pass

文档增强

class DocumentedModel(BaseModel):
    """用户模型文档说明"""
    name: str = Field(..., description="用户姓名", example="张三")
    age: int = Field(description="年龄范围18-100", example=25)

性能监控

import time
from pydantic import BaseModel

class BenchmarkModel(BaseModel):
    def __init__(self, **data):
        start = time.perf_counter()
        super().__init__(**data)
        print(f"模型初始化耗时: {time.perf_counter()-start:.4f}s")

参考链接：

• Welcome to Pydantic – Pydantic
• GitHub – pydantic/pydantic: Data validation using Python type hints