FastAPI: 请求体与数据验证 — Pydantic V2 深度实践
最后更新:2026-08-26
数据验证就像机场安检——每个乘客(请求)必须通过身份核验(类型检查)、行李扫描(约束校验)和申报审核(自定义验证器),才能登机(进入业务逻辑)。
1. 你将学到
- Pydantic V2
BaseModel:field_validator/model_validator替代 V1 的@validator Field()高级约束:gt/lt/pattern/examples与 JSON Schema 生成- 嵌套模型与模型组合:
Optional、Union、Literal的实战用法 model_config:ConfigDict替代 V1 的class Config,from_attributes=TrueORM 模式- Alice 场景:PriceTracker 商品价格提交的完整 Pydantic 模型设计
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) 完整生命周期
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 迁移对照
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 倍。百万级数据场景提升明显。
📖 小节
- Pydantic V2 用
@field_validator和@model_validator替代 V1 的@validator和@root_validator Field()约束自动映射为 JSON Schema,同时驱动 OpenAPI 文档和请求数据校验- 嵌套模型支持
Optional、Union、Literal灵活组合,Literal实现判别联合类型 ConfigDict(from_attributes=True)启用 ORM 模式,从 SQLAlchemy 对象直接创建 Pydantic 模型- V1→V2 迁移核心:
.dict()→.model_dump(),class Config→model_config = ConfigDict(...)
📝 作业
- 基础题(难度⭐):创建
PriceCreate模型,包含product_id: int(> 0)和price: float(> 0),验证非法输入抛出校验错误。提示:BaseModel+Field(gt=0) - 进阶题(难度⭐⭐):为
PriceCreate添加@field_validator校验currency字段只能是 USD/EUR/GBP,并添加@model_validator确保min_price <= max_price。提示:field_validator+model_validator(mode="after") - 挑战题(难度⭐⭐⭐):设计 PriceTracker 完整的嵌套模型体系:
ProductCreate内嵌PriceInfo(amount + currency),PriceInfo的 currency 用正则约束,ProductCreate的 SKU 用pattern约束格式。提示:Field(pattern=...)+ 嵌套BaseModel
---|