FastAPI: 请求体与数据验证 — Pydantic V2 深度实践

最后更新:2026-08-26

数据验证就像机场安检——每个乘客(请求)必须通过身份核验(类型检查)、行李扫描(约束校验)和申报审核(自定义验证器),才能登机(进入业务逻辑)。

1. 你将学到


2. Alice 的真实故事

(1) 痛点:价格数据提交校验漏洞百出

Alice 的 PriceTracker 接收供应商提交的价格数据,要求价格必须 > 0、币种为 ISO 4217 三字母代码、商品名称不能为空。但 Bob 前端偶尔发送负数价格或非法币种,Flask 手写校验代码散落在 10 个地方,漏掉了"价格为 0"的边界情况,导致数据库出现价格为 0 的脏数据。

(2) Pydantic V2 声明式校验的解法

Pydantic V2 用声明式模型替代命令式校验,所有约束写在模型定义中,一处定义全局生效。

PYTHON
from pydantic import BaseModel, Field, field_validator

class PriceCreate(BaseModel):
    product_id: int = Field(gt=0)
    price: float = Field(gt=0, description="Price in USD")
    currency: str = Field(pattern=r"^[A-Z]{3}$", examples=["USD", "EUR"])

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        if v not in {"USD", "EUR", "GBP", "JPY", "CNY"}:
            raise ValueError(f"Unsupported currency: {v}")
        return v

(3) 收益

价格校验代码从 10 处分散逻辑集中到 1 个模型定义,0 值和负数价格自动被拦截,币种校验由正则+自定义验证器双重保障,数据库再无脏数据。


3. Pydantic V2 数据流

(1) 完整生命周期

100%
flowchart TD
    A[JSON Request Body] --> B[model_validate]
    B --> C[Type Coercion]
    C --> D[field_validator]
    D --> E[model_validator]
    E --> F[Valid Model Instance]
    F --> G[model_dump]
    G --> H[JSON Response]
    F --> I[model_dump_json]
    I --> J[JSON String]
阶段 方法 说明
输入解析 model_validate(data) 从 dict/JSON 解析并校验
类型转换 自动 "42"42"9.99"9.99
字段校验 @field_validator 单字段自定义校验
模型校验 @model_validator 跨字段联合校验
输出序列化 model_dump() / model_dump_json() 模型转 dict/JSON

▶ 示例:Pydantic V2 基础模型

PYTHON
from pydantic import BaseModel, Field

class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    category: str = Field(max_length=100)
    base_price: float = Field(gt=0, description="Base price in USD")

# Parse and validate
data = {"name": "Widget", "category": "electronics", "base_price": 29.99}
product = ProductCreate.model_validate(data)
print(product.model_dump())

输出:

TEXT 📖 仅展示
{'name': 'Widget', 'category': 'electronics', 'base_price': 29.99}

4. field_validator 与 model_validator

(1) V1 → V2 迁移对照

100%
flowchart LR
    V1[Pydantic V1] --> Migrate[V1 → V2 Migration]
    Migrate --> V2[Pydantic V2]
    V1 --- A["@validator('field')"]
    V2 --- B["@field_validator('field')"]
    V1 --- C["@root_validator"]
    V2 --- D["@model_validator(mode='after')"]
    V1 --- E["class Config:"]
    V2 --- F["model_config = ConfigDict(...)"]
    V1 --- G[".dict()"]
    V2 --- H[".model_dump()"]
V1 写法 V2 写法 说明
@validator("field") @field_validator("field") 字段校验器
@root_validator @model_validator(mode="after") 模型级校验
class Config: orm_mode = True model_config = ConfigDict(from_attributes=True) ORM 模式
.dict() .model_dump() 序列化为 dict
.json() .model_dump_json() 序列化为 JSON

▶ 示例:field_validator 单字段校验

PYTHON
from pydantic import BaseModel, Field, field_validator

class PriceCreate(BaseModel):
    product_id: int = Field(gt=0)
    price: float = Field(gt=0)
    currency: str = Field(default="USD", max_length=3)

    @field_validator("price")
    @classmethod
    def price_precision(cls, v: float) -> float:
        # Round to 2 decimal places
        return round(v, 2)

    @field_validator("currency")
    @classmethod
    def valid_currency(cls, v: str) -> str:
        allowed = {"USD", "EUR", "GBP", "JPY", "CNY"}
        if v not in allowed:
            raise ValueError(f"Currency must be one of {allowed}")
        return v.upper()

# Test validation
p = PriceCreate(product_id=1, price=9.999, currency="usd")
print(p.model_dump())

输出:

TEXT 📖 仅展示
{'product_id': 1, 'price': 10.0, 'currency': 'USD'}

▶ 示例:model_validator 跨字段校验

PYTHON
from pydantic import BaseModel, Field, model_validator

class PriceRangeQuery(BaseModel):
    min_price: float = Field(ge=0, description="Min price in USD")
    max_price: float = Field(ge=0, description="Max price in USD")

    @model_validator(mode="after")
    def check_range(self):
        if self.min_price > self.max_price:
            raise ValueError("min_price must be <= max_price")
        return self

# Valid
valid = PriceRangeQuery(min_price=10, max_price=100)
print(valid.model_dump())

# Invalid - raises validation error
# PriceRangeQuery(min_price=100, max_price=10)

输出:

TEXT 📖 仅展示
{'min_price': 10.0, 'max_price': 100.0}

5. Field() 高级约束与 JSON Schema

(1) Field() 参数速查

参数 类型 说明 JSON Schema 映射
gt 数值 大于 exclusiveMinimum
ge 数值 大于等于 minimum
lt 数值 小于 exclusiveMaximum
le 数值 小于等于 maximum
min_length 字符串 最小长度 minLength
max_length 字符串 最大长度 maxLength
pattern 字符串 正则 pattern
default 任意 默认值 default
examples 列表 示例值 examples
description 字符串 描述 description
alias 字符串 字段别名 别名映射

▶ 示例:Field 约束与 JSON Schema

PYTHON
from pydantic import BaseModel, Field

class ProductCreate(BaseModel):
    name: str = Field(
        min_length=1,
        max_length=200,
        description="Product display name",
        examples=["Wireless Mouse", "USB Cable"],
    )
    sku: str = Field(
        pattern=r"^[A-Z]{2}-\d{4,6}$",
        description="SKU code: 2 letters + 4-6 digits",
        examples=["EL-1234", "CB-567890"],
    )
    base_price: float = Field(
        gt=0,
        le=999999.99,
        description="Base price in USD",
        examples=[9.99, 49.99, 199.99],
    )

# View generated JSON Schema
print(ProductCreate.model_json_schema())

输出:

TEXT 📖 仅展示
# 执行成功

6. 嵌套模型与模型组合

(1) 模型嵌套结构

▶ 示例:PriceTracker 的嵌套模型

PYTHON
from pydantic import BaseModel, Field
from typing import Optional

class PriceInfo(BaseModel):
    amount: float = Field(gt=0, description="Price amount in USD")
    currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")
    source: str = Field(max_length=100, description="Price source")

class ProductCreate(BaseModel):
    name: str = Field(min_length=1, max_length=200)
    category: str = Field(max_length=100)
    current_price: PriceInfo  # Nested model
    original_price: Optional[PriceInfo] = None  # Optional nested

# Nested validation
data = {
    "name": "Wireless Mouse",
    "category": "electronics",
    "current_price": {"amount": 29.99, "currency": "USD", "source": "Amazon"},
    "original_price": {"amount": 49.99, "currency": "USD", "source": "Amazon"},
}
product = ProductCreate.model_validate(data)
print(product.model_dump())

输出:

TEXT 📖 仅展示
# 执行成功

▶ 示例:Union 和 Literal

PYTHON
from pydantic import BaseModel, Field
from typing import Union, Literal

class SinglePrice(BaseModel):
    type: Literal["single"] = "single"
    amount: float = Field(gt=0)

class RangePrice(BaseModel):
    type: Literal["range"] = "range"
    min_amount: float = Field(gt=0)
    max_amount: float = Field(gt=0)

class ProductPrice(BaseModel):
    product_id: int = Field(gt=0)
    pricing: Union[SinglePrice, RangePrice]  # Discriminated union

# FastAPI uses "type" field to determine which model to validate
data = {"product_id": 1, "pricing": {"type": "range", "min_amount": 10, "max_amount": 50}}
pp = ProductPrice.model_validate(data)
print(pp.model_dump())

输出:

TEXT 📖 仅展示
# 执行成功

(2) ConfigDict 配置

▶ 示例:model_config 与 ORM 模式

PYTHON
from pydantic import BaseModel, ConfigDict

class ProductResponse(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,  # Enable ORM mode (read from SQLAlchemy objects)
        populate_by_name=True,  # Allow both field name and alias
        json_schema_extra={
            "examples": [{"id": 1, "name": "Widget", "price": 9.99}]
        },
    )

    id: int
    name: str
    price: float

# With from_attributes=True, can create from object attributes
class FakeORMObject:
    def __init__(self):
        self.id = 1
        self.name = "Widget"
        self.price = 9.99

orm_obj = FakeORMObject()
response = ProductResponse.model_validate(orm_obj)
print(response.model_dump())

输出:

TEXT 📖 仅展示
{'id': 1, 'name': 'Widget', 'price': 9.99}

7. 综合示例

Pydantic V2 的字段校验、跨字段验证与 ORM 模式配合 FastAPI 请求体,实现完整的数据输入校验链路。

PYTHON
from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator, model_validator, ConfigDict

class PriceCreate(BaseModel):
    product_name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0, description="价格必须大于 0")
    currency: str = "USD"

    @field_validator("currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        if v not in ("USD", "EUR", "GBP"):
            raise ValueError("currency must be USD/EUR/GBP")
        return v

class PriceRangeQuery(BaseModel):
    min_price: float = Field(ge=0)
    max_price: float = Field(ge=0)

    @model_validator(mode="after")
    def validate_range(self):
        if self.min_price > self.max_price:
            raise ValueError("min_price must <= max_price")
        return self

app = FastAPI()

@app.post("/prices")
async def create_price(data: PriceCreate):
    return data.model_dump()

输出:

TEXT 📖 仅展示
POST /prices {"product_name":"Widget","price":9.99} → {"product_name":"Widget","price":9.99,"currency":"USD"}
POST /prices {"product_name":"","price":-1} → 422 Validation Error

❓ 常见问题

Q field_validator 和 model_validator 怎么选?
A 单字段校验用 field_validator(如格式检查),跨字段联合校验用 model_validator(如 min_price <= max_price)。
Q V1 的 @validator 还能用吗?
A V2 保留了兼容层但会发出废弃警告。新项目必须用 @field_validator/@model_validator,旧项目尽快迁移。
Q from_attributes=True 有什么用?
A 允许从 ORM 对象(如 SQLAlchemy model 实例)直接创建 Pydantic 模型,读取对象属性而非 dict。这是 FastAPI + SQLAlchemy 的关键配置。
Q Field() 的 examples 和 json_schema_extra 有什么区别?
A examples 是字段的示例值列表,映射到 OpenAPI examples;json_schema_extra 是模型级额外 Schema 属性。
Q 嵌套模型过深有什么问题?
A 嵌套超过 3 层会增加校验延迟和调试难度。建议扁平化或使用组合模式拆分。
Q Pydantic V2 的性能比 V1 好多少?
A 核心校验快 5-50 倍(Rust 实现),序列化快 2-10 倍。百万级数据场景提升明显。

📖 小节


📝 作业

  1. 基础题(难度⭐):创建 PriceCreate 模型,包含 product_id: int(> 0)和 price: float(> 0),验证非法输入抛出校验错误。提示:BaseModel + Field(gt=0)
  2. 进阶题(难度⭐⭐):为 PriceCreate 添加 @field_validator 校验 currency 字段只能是 USD/EUR/GBP,并添加 @model_validator 确保 min_price <= max_price。提示:field_validator + model_validator(mode="after")
  3. 挑战题(难度⭐⭐⭐):设计 PriceTracker 完整的嵌套模型体系:ProductCreate 内嵌 PriceInfo(amount + currency),PriceInfo 的 currency 用正则约束,ProductCreate 的 SKU 用 pattern 约束格式。提示:Field(pattern=...) + 嵌套 BaseModel

---|

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核,确保内容准确可靠。如发现任何问题,欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站,几个人的小团队,精力有限。页面虽经检查,难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了,麻烦告诉我们,我们会在收到反馈后第一时间进行修复,再次感谢您的光临 🙏