Python Pydantic V2資料校驗實戰:從模型設計到自定義校驗器的7種生產模式
编程语言
Pydantic V2:你的資料校驗還在手寫if-else嗎?
介面引數校驗漏了欄位,資料庫寫入了髒資料,設定檔解析出了None——這些生產事故的根因都是資料校驗不嚴格。你手寫if-else校驗,程式碼又臭又長還容易漏;你用V1的@validator,遷移到V2發現全報錯;你配置了model_config,序列化結果還是不對。2026年,Pydantic V2已經全面取代V1,效能提升5-50倍,但API變化巨大,遷移坑多。
本文將從7種生產模式出發,帶你完成基礎模型→欄位校驗→自定義校驗器→序列化→JSON Schema→效能最佳化→FastAPI整合的全鏈路實戰,每一步都有完整程式碼和避坑指南。
Pydantic V2核心概念
| 概念 | 說明 |
|---|---|
| BaseModel | Pydantic核心類,定義資料模型並自動校驗 |
| Field | 欄位配置,支援預設值、描述、約束條件 |
| field_validator | V2新增欄位校驗器,替代V1的@validator |
| model_validator | 模型級校驗器,跨欄位聯合校驗 |
| model_config | 模型配置,控制序列化、嚴格模式等行為 |
| TypeAdapter | 非BaseModel型別的校驗適配器 |
| JSON Schema | 從模型自動生成JSON Schema,用於API文件 |
| Serialize | 序列化控制,支援exclude、alias、自定義序列化 |
問題分析:資料校驗的5大痛點
- 手寫校驗程式碼冗長易錯:每個介面寫一堆if-else,漏了欄位就出bug,維護成本高
- V1到V2遷移API不相容:
@validator變成@field_validator,Config類變成model_config,大量程式碼要改 - 巢狀模型序列化失控:ORM物件轉JSON時迴圈引用、敏感欄位洩露、欄位名不符合前端約定
- 跨欄位聯合校驗難實現:密碼確認、日期範圍、條件必填等場景需要多個欄位一起校驗
- 效能瓶頸:V1在大資料量下校驗慢,V2雖然快了但配置不當反而更慢
分步實操:7種Pydantic V2生產模式
模式1:基礎模型設計與欄位約束
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime
from enum import Enum
class UserStatus(str, Enum):
ACTIVE = "active"
INACTIVE = "inactive"
SUSPENDED = "suspended"
class UserCreate(BaseModel):
model_config = {"str_strip_whitespace": True, "str_min_length": 1}
username: str = Field(
min_length=3,
max_length=20,
pattern=r"^[a-zA-Z0-9_]+$",
description="使用者名稱,3-20位字母數字底線"
)
email: EmailStr = Field(description="郵箱地址")
password: str = Field(
min_length=8,
max_length=128,
description="密碼,8-128位"
)
age: Optional[int] = Field(
default=None,
ge=0,
le=150,
description="年齡,0-150"
)
status: UserStatus = Field(default=UserStatus.ACTIVE)
created_at: datetime = Field(default_factory=datetime.now)
class UserResponse(BaseModel):
id: int = Field(gt=0)
username: str
email: EmailStr
status: UserStatus
created_at: datetime
user = UserCreate(
username="zhang_san",
email="zhang@example.com",
password="secureP@ss123",
age=28
)
print(user.model_dump())
模式2:欄位級校驗器field_validator
from pydantic import BaseModel, Field, field_validator
import re
class RegisterRequest(BaseModel):
username: str = Field(min_length=3, max_length=20)
password: str = Field(min_length=8)
confirm_password: str
@field_validator("username")
@classmethod
def username_must_be_valid(cls, v: str) -> str:
if not re.match(r"^[a-zA-Z0-9_]+$", v):
raise ValueError("使用者名稱只能包含字母、數字和底線")
if v.startswith("_"):
raise ValueError("使用者名稱不能以底線開頭")
return v.lower()
@field_validator("password")
@classmethod
def password_strength_check(cls, v: str) -> str:
if not re.search(r"[A-Z]", v):
raise ValueError("密碼必須包含至少一個大寫字母")
if not re.search(r"[a-z]", v):
raise ValueError("密碼必須包含至少一個小寫字母")
if not re.search(r"\d", v):
raise ValueError("密碼必須包含至少一個數字")
if not re.search(r"[!@#$%^&*(),.?\":{}|<>]", v):
raise ValueError("密碼必須包含至少一個特殊字元")
return v
class ProductCreate(BaseModel):
name: str = Field(min_length=1, max_length=200)
price: float = Field(gt=0)
tags: list[str] = Field(default_factory=list)
@field_validator("tags")
@classmethod
def tags_deduplicate(cls, v: list[str]) -> list[str]:
seen = set()
result = []
for tag in v:
tag_lower = tag.lower().strip()
if tag_lower and tag_lower not in seen:
seen.add(tag_lower)
result.append(tag_lower)
return result
@field_validator("price")
@classmethod
def price_round_to_cents(cls, v: float) -> float:
return round(v, 2)
模式3:模型級校驗器model_validator
from pydantic import BaseModel, Field, model_validator
from datetime import date, timedelta
from typing import Optional
class DateRangeQuery(BaseModel):
start_date: date
end_date: date
@model_validator(mode="after")
def validate_date_range(self) -> "DateRangeQuery":
if self.start_date > self.end_date:
raise ValueError("開始日期不能晚於結束日期")
if (self.end_date - self.start_date).days > 365:
raise ValueError("查詢範圍不能超過365天")
return self
class EventCreate(BaseModel):
title: str = Field(min_length=1, max_length=200)
event_type: str
start_time: datetime
end_time: Optional[datetime] = None
location: Optional[str] = None
online_url: Optional[str] = None
@model_validator(mode="after")
def validate_event(self) -> "EventCreate":
if self.event_type == "offline" and not self.location:
raise ValueError("線下活動必須填寫地點")
if self.event_type == "online" and not self.online_url:
raise ValueError("線上活動必須填寫連結")
if self.event_type == "hybrid":
if not self.location:
raise ValueError("混合活動必須填寫線下地點")
if not self.online_url:
raise ValueError("混合活動必須填寫線上連結")
if self.end_time and self.start_time >= self.end_time:
raise ValueError("結束時間必須晚於開始時間")
return self
class PasswordChange(BaseModel):
old_password: str = Field(min_length=1)
new_password: str = Field(min_length=8)
confirm_password: str
@model_validator(mode="after")
def passwords_match(self) -> "PasswordChange":
if self.new_password != self.confirm_password:
raise ValueError("兩次輸入的新密碼不一致")
if self.old_password == self.new_password:
raise ValueError("新密碼不能與舊密碼相同")
return self
模式4:序列化控制與別名
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional
class UserORM(BaseModel):
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True,
)
id: int
username: str = Field(alias="user_name")
email: str = Field(alias="email_address")
hashed_password: str = Field(exclude=True)
phone: Optional[str] = Field(default=None, exclude=True)
avatar_url: Optional[str] = Field(default=None, serialization_alias="avatar")
created_at: datetime
updated_at: Optional[datetime] = None
class ArticleResponse(BaseModel):
model_config = ConfigDict(populate_by_name=True)
id: int
title: str
content: str = Field(exclude=True)
summary: Optional[str] = None
author_id: int = Field(serialization_alias="authorId")
tags: list[str] = Field(default_factory=list)
view_count: int = Field(default=0, serialization_alias="viewCount")
created_at: datetime = Field(serialization_alias="createdAt")
updated_at: Optional[datetime] = Field(default=None, serialization_alias="updatedAt")
def get_summary(self) -> str:
if self.summary:
return self.summary
return self.content[:200] + "..." if len(self.content) > 200 else self.content
article = ArticleResponse(
id=1,
title="Pydantic V2實戰指南",
content="這是一篇很長的文章內容..." * 50,
author_id=42,
tags=["Python", "Pydantic"],
view_count=1024,
created_at=datetime.now()
)
print(article.model_dump(by_alias=True))
模式5:JSON Schema生成與API文件
from pydantic import BaseModel, Field
import json
class APIRequest(BaseModel):
"""建立訂單請求"""
product_id: int = Field(gt=0, description="商品ID")
quantity: int = Field(ge=1, le=999, description="購買數量")
coupon_code: Optional[str] = Field(default=None, pattern=r"^[A-Z0-9]{6,12}$", description="優惠券碼")
shipping_address: str = Field(min_length=5, max_length=500, description="收貨地址")
remark: Optional[str] = Field(default=None, max_length=200, description="訂單備註")
class APIResponse(BaseModel):
"""建立訂單響應"""
order_id: str = Field(description="訂單號")
total_amount: float = Field(description="訂單總金額")
discount_amount: float = Field(default=0.0, description="優惠金額")
final_amount: float = Field(description="實付金額")
status: str = Field(description="訂單狀態")
schema = APIRequest.model_json_schema()
print(json.dumps(schema, indent=2, ensure_ascii=False))
模式6:TypeAdapter與泛型校驗
from pydantic import BaseModel, TypeAdapter, Field
from typing import Generic, TypeVar, Optional
T = TypeVar("T")
class PageResponse(BaseModel, Generic[T]):
items: list[T]
total: int = Field(ge=0)
page: int = Field(ge=1)
page_size: int = Field(ge=1, le=100)
has_next: bool
class UserItem(BaseModel):
id: int
username: str
email: str
user_page_type = PageResponse[UserItem]
adapter = TypeAdapter(user_page_type)
json_data = {
"items": [
{"id": 1, "username": "alice", "email": "alice@example.com"},
{"id": 2, "username": "bob", "email": "bob@example.com"},
],
"total": 100,
"page": 1,
"page_size": 10,
"has_next": True
}
page = adapter.validate_python(json_data)
print(page.model_dump())
raw_list_adapter = TypeAdapter(list[int])
result = raw_list_adapter.validate_python(["1", "2", "3"])
print(result)
config_adapter = TypeAdapter(dict[str, int])
config = config_adapter.validate_python({"timeout": "30", "retries": "3"})
print(config)
模式7:FastAPI整合生產實踐
from fastapi import FastAPI, HTTPException, Depends, Query
from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Optional
app = FastAPI(title="使用者管理API")
class UserCreateRequest(BaseModel):
username: str = Field(min_length=3, max_length=20, pattern=r"^[a-zA-Z0-9_]+$")
email: str = Field(pattern=r"^[\w.-]+@[\w.-]+\.\w+$")
password: str = Field(min_length=8, max_length=128)
role: str = Field(default="user", pattern=r"^(admin|user|guest)$")
@field_validator("password")
@classmethod
def password_strength(cls, v: str) -> str:
has_upper = any(c.isupper() for c in v)
has_lower = any(c.islower() for c in v)
has_digit = any(c.isdigit() for c in v)
if not (has_upper and has_lower and has_digit):
raise ValueError("密碼必須包含大寫字母、小寫字母和數字")
return v
class UserUpdateRequest(BaseModel):
email: Optional[str] = None
role: Optional[str] = None
status: Optional[str] = None
@model_validator(mode="after")
def at_least_one_field(self) -> "UserUpdateRequest":
if self.email is None and self.role is None and self.status is None:
raise ValueError("至少需要更新一個欄位")
return self
class UserDetailResponse(BaseModel):
id: int
username: str
email: str
role: str
status: str
created_at: datetime
class ErrorResponse(BaseModel):
error_code: int
message: str
detail: Optional[str] = None
@app.post("/users", response_model=UserDetailResponse, responses={400: {"model": ErrorResponse}})
async def create_user(req: UserCreateRequest):
user_data = req.model_dump()
user_data["id"] = 1
user_data["status"] = "active"
user_data["created_at"] = datetime.now()
return user_data
@app.patch("/users/{user_id}", response_model=UserDetailResponse)
async def update_user(user_id: int, req: UserUpdateRequest):
update_data = req.model_dump(exclude_none=True)
if not update_data:
raise HTTPException(status_code=400, detail="No fields to update")
return {"id": user_id, "username": "test", "email": "test@example.com", "role": "user", "status": "active", "created_at": datetime.now()}
@app.get("/users", response_model=PageResponse[UserDetailResponse])
async def list_users(
page: int = Query(ge=1, default=1),
page_size: int = Query(ge=1, le=100, default=20),
role: Optional[str] = Query(default=None, pattern=r"^(admin|user|guest)$"),
):
return {
"items": [],
"total": 0,
"page": page,
"page_size": page_size,
"has_next": False
}
避坑指南
坑1:V1的@validator直接改成@field_validator不生效
# ❌ 錯誤:V1寫法直接改名,缺少cls和mode引數
from pydantic import field_validator
class Bad(BaseModel):
name: str
@field_validator("name")
def validate_name(v):
return v.upper()
# ✅ 正確:V2必須加@classmethod和mode引數
class Good(BaseModel):
name: str
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
return v.upper()
坑2:model_config寫成內部類
# ❌ 錯誤:V1的Config內部類寫法,V2已廢棄
class OldWay(BaseModel):
name: str
class Config:
orm_mode = True
# ✅ 正確:V2使用model_config字典
class NewWay(BaseModel):
model_config = {"from_attributes": True}
name: str
# ✅ 更好:使用ConfigDict獲得型別提示
from pydantic import ConfigDict
class BestWay(BaseModel):
model_config = ConfigDict(from_attributes=True)
name: str
坑3:序列化時exclude不生效
class User(BaseModel):
id: int
name: str
password: str = Field(exclude=True)
user = User(id=1, name="test", password="secret")
# ❌ 錯誤:model_dump()預設不應用序列化別名
print(user.model_dump())
# {'id': 1, 'name': 'test', 'password': 'secret'} # password還在!
# ✅ 正確:需要加mode引數
print(user.model_dump(mode="python"))
# {'id': 1, 'name': 'test'} # password被排除
# ✅ JSON序列化
print(user.model_dump_json())
# {"id":1,"name":"test"} # password被排除
坑4:from_attributes與ORM欄位不匹配
# ❌ 錯誤:ORM欄位名與模型欄位名不一致,from_attributes靜默跳過
class ORMUser:
def __init__(self):
self.user_name = "test" # ORM欄位名
self.email_addr = "t@e.com"
class PydanticUser(BaseModel):
model_config = ConfigDict(from_attributes=True)
username: str # 不匹配user_name
email: str # 不匹配email_addr
# ✅ 正確:使用Field(alias=...)對映ORM欄位名
class PydanticUserFixed(BaseModel):
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
username: str = Field(alias="user_name")
email: str = Field(alias="email_addr")
坑5:Optional欄位傳None不校驗
# ❌ 錯誤:Optional欄位傳None跳過了校驗
class Bad(BaseModel):
age: Optional[int] = Field(None, ge=0, le=150)
Bad(age=None) # 透過,但None不是合法年齡
# ✅ 正確:區分"可選"和"允許None",用顯式Union
from typing import Union
class Good(BaseModel):
age: Union[int, None] = Field(None, ge=0, le=150)
# ✅ 更好:如果None有意義,用自定義校驗器處理
class Better(BaseModel):
age: Optional[int] = Field(None, ge=0, le=150)
@field_validator("age")
@classmethod
def age_not_none_if_provided(cls, v: Optional[int]) -> Optional[int]:
if v is not None and v < 0:
raise ValueError("年齡不能為負數")
return v
報錯排查
| 序號 | 報錯資訊 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | ValidationError: field required |
必填欄位未提供 | 檢查欄位是否有default或default_factory |
| 2 | ValidationError: string too short |
字串長度不足 | 調整min_length或輸入更長的值 |
| 3 | PydanticUserWarning: @validator is deprecated |
使用了V1的@validator | 替換為@field_validator並加@classmethod |
| 4 | AttributeError: 'Config' class not supported |
V2不支援內部Config類 | 改用model_config字典或ConfigDict |
| 5 | ValidationError: Input should be a valid integer |
型別轉換失敗 | 檢查輸入是否為合法數字字串 |
| 6 | ValueError: field_validator missing cls |
field_validator缺少@classmethod | 在@field_validator下方新增@classmethod |
| 7 | ValidationError: Extra inputs are not permitted |
嚴格模式下多餘欄位被拒絕 | 設定model_config的extra="ignore"或"allow" |
| 8 | TypeError: Unable to generate pydantic-core schema |
型別註解不被支援 | 檢查是否使用了複雜泛型或未支援的型別 |
| 9 | RecursionError: maximum recursion depth exceeded |
巢狀模型迴圈引用 | 使用Optional前向引用或重構模型 |
| 10 | SerializationError: circular reference detected |
序列化時檢測到迴圈引用 | 使用exclude引數或自定義序列化器 |
進階最佳化
1. 嚴格模式與寬鬆模式切換
from pydantic import BaseModel, ConfigDict, StrictInt, StrictStr
class StrictModel(BaseModel):
model_config = ConfigDict(strict=True)
id: int
name: str
class LaxModel(BaseModel):
model_config = ConfigDict(strict=False)
id: int
name: str
strict_result = StrictModel(id=1, name="test")
lax_result = LaxModel(id="1", name="test")
class HybridModel(BaseModel):
model_config = ConfigDict(strict=False)
id: StrictInt
name: str
2. 自定義型別與Annotated
from pydantic import BaseModel, BeforeValidator, AfterValidator
from typing import Annotated
def normalize_phone(v: str) -> str:
return v.replace("-", "").replace(" ", "").replace("+86", "")
def check_phone_format(v: str) -> str:
if not v.startswith("1") or len(v) != 11:
raise ValueError("手機號格式不正確")
return v
PhoneNumber = Annotated[str, BeforeValidator(normalize_phone), AfterValidator(check_phone_format)]
YuanFromCents = Annotated[float, BeforeValidator(lambda v: v / 100 if isinstance(v, int) else v)]
class PaymentRequest(BaseModel):
phone: PhoneNumber
amount: YuanFromCents = Field(gt=0, description="金額(元)")
payment = PaymentRequest(phone="+86-138-0013-8000", amount=9900)
print(payment.model_dump())
3. 效能最佳化:快取與預編譯
from pydantic import BaseModel, TypeAdapter
import time
class LargeModel(BaseModel):
field1: str
field2: int
field3: float
field4: bool
field5: str
field6: int
field7: float
field8: bool
adapter = TypeAdapter(LargeModel)
data = {"field1": "a", "field2": 1, "field3": 1.0, "field4": True, "field5": "b", "field6": 2, "field7": 2.0, "field8": False}
start = time.perf_counter()
for _ in range(100000):
LargeModel(**data)
v1_time = time.perf_counter() - start
start = time.perf_counter()
for _ in range(100000):
adapter.validate_python(data)
adapter_time = time.perf_counter() - start
print(f"Direct: {v1_time:.3f}s, TypeAdapter: {adapter_time:.3f}s")
對比分析
| 維度 | Pydantic V1 | Pydantic V2 | 手寫if-else | Marshmallow |
|---|---|---|---|---|
| 校驗效能 | ⭐⭐慢 | ⭐⭐⭐⭐⭐快5-50x | ⭐⭐⭐⭐快 | ⭐⭐慢 |
| 型別提示整合 | ⚠️部分 | ✅完整 | ❌無 | ❌無 |
| 錯誤資訊 | ⚠️一般 | ✅詳細定位 | ❌自定義 | ⚠️一般 |
| JSON Schema | ✅支援 | ✅完善 | ❌無 | ✅支援 |
| 序列化控制 | ⚠️有限 | ✅靈活 | ❌手寫 | ✅靈活 |
| 學習曲線 | ⭐⭐低 | ⭐⭐⭐中 | ⭐最低 | ⭐⭐⭐中 |
| FastAPI整合 | ✅原生 | ✅原生 | ❌無 | ⚠️需適配 |
| 生產推薦 | 遺留專案 | 首選 | 簡單指令碼 | 複雜轉換 |
總結:Pydantic V2不是簡單的版本升級,而是從「校驗庫」到「資料工程基礎設施」的質變。核心原則三條:用Field約束代替手寫校驗、用model_validator處理跨欄位邏輯、用model_config控制序列化行為。V1到V2的遷移雖然痛苦,但5-50倍的效能提升和更完善的型別系統值得投入。FastAPI + Pydantic V2已經是2026年Python Web開發的事實標準。
線上工具推薦
- JSON格式化:/zh-TW/json/format
- Base64編解碼:/zh-TW/encode/base64
- Hash計算:/zh-TW/encode/hash
- JWT解碼:/zh-TW/encode/jwt-decode
本站提供瀏覽器本地工具,免註冊即可試用 →
#Python#Pydantic#数据校验#FastAPI#类型注解#2026#JSON Schema