当前位置:首页>python>Python Web开发:Pydantic进阶

Python Web开发:Pydantic进阶

  • 2026-10-11 07:14:35
Python Web开发:Pydantic进阶
副标题

: 90%的人不知道,Pydantic v2的性能比v1快10倍

痛点:为什么你的数据验证代码总是很繁琐?

2025年某项目有100+数据模型,每个都要写一堆验证逻辑。问题出在哪?工程师没有使用Pydantic的高级特性。

真相

:Pydantic v2用Rust重写,性能提升10倍,功能更强大。

特性v1v2提升
性能1x10x10倍
类型检查基础完整更准确
序列化慢快5-10倍
异步支持无有新增

一、Pydantic v2基础

1.1 安装与升级

# 安装Pydantic v2

pip install pydantic>=2.0

检查版本

python -c "import pydantic; print(pydantic.__version__)"

1.2 基础模型

from pydantic import BaseModel, Field

class User(BaseModel):

id: int

username: str = Field(..., min_length=3, max_length=50)

email: str

age: int = Field(..., gt=0, lt=150)

is_active: bool = True

class Config:

from_attributes = True # v2中用model_config替代

v2使用model_config

class UserV2(BaseModel):

model_config = ConfigDict(from_attributes=True)

id: int

username: str

email: str

1.3 数据验证

from pydantic import BaseModel, Field, field_validator, model_validator

from typing import Annotated

from datetime import datetime

class Product(BaseModel):

name: str = Field(..., min_length=1, max_length=100)

price: Annotated[float, Field(gt=0)]

quantity: int = Field(ge=0)

category: str

@field_validator('name')

@classmethod

def validate_name(cls, v: str) -> str:

if not v.strip():

raise ValueError('名称不能为空')

return v.strip().title()

@field_validator('price')

@classmethod

def validate_price(cls, v: float) -> float:

return round(v, 2)

@model_validator(mode='after')

def validate_stock(self):

if self.price > 1000 and self.quantity > 100:

raise ValueError('高价商品库存不能过多')

return self

二、字段类型

2.1 基础类型

from pydantic import BaseModel

from typing import Any

class Types(BaseModel):

# 基础类型

integer: int

float_num: float

string: str

boolean: bool

# 可空类型

optional_int: int | None = None

# 默认值

default_str: str = "default"

default_list: list[int] = [1, 2, 3]

2.2 容器类型

from pydantic import BaseModel

from typing import List, Dict, Set, Tuple

class Collections(BaseModel):

# 列表

items: List[str]

# 字典

metadata: Dict[str, Any]

# 集合(自动去重)

tags: Set[str]

# 元组

coordinates: Tuple[float, float]

# 可变长度

numbers: list[int]

2.3 联合类型

from pydantic import BaseModel

from typing import Union

class Flexible(BaseModel):

# 联合类型

value: Union[int, str, float]

# Python 3.10+语法

value_v2: int | str | float

# 可选联合

optional: int | None = None

三、高级验证

3.1 模式验证

from pydantic import BaseModel, Field, field_validator

import re

class User(BaseModel):

username: str = Field(..., pattern=r'^[a-zA-Z0-9_]{3,20}$')

phone: str = Field(..., pattern=r'^1[3-9]\d{9}$')

email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')

@field_validator('username')

@classmethod

def validate_username(cls, v: str) -> str:

if v.lower() in ['admin', 'root', 'system']:

raise ValueError('用户名不能是保留字')

return v

3.2 自定义验证器

from pydantic import BaseModel, field_validator, model_validator

from datetime import datetime

class Event(BaseModel):

name: str

start_time: datetime

end_time: datetime

max_attendees: int = 100

@field_validator('start_time', 'end_time')

@classmethod

def validate_datetime(cls, v: datetime) -> datetime:

if v < datetime.now():

raise ValueError('时间不能是过去')

return v

@model_validator(mode='after')

def validate_time_range(self):

if self.end_time <= self.start_time:

raise ValueError('结束时间必须晚于开始时间')

return self

3.3 条件验证

from pydantic import BaseModel, Field, field_validator

from typing import Optional

class Payment(BaseModel):

amount: float

method: str

card_number: Optional[str] = None

paypal_email: Optional[str] = None

@field_validator('card_number')

@classmethod

def validate_card(cls, v: Optional[str], info) -> Optional[str]:

if info.data.get('method') == 'card' and not v:

raise ValueError('信用卡支付需要卡号')

return v

@field_validator('paypal_email')

@classmethod

def validate_paypal(cls, v: Optional[str], info) -> Optional[str]:

if info.data.get('method') == 'paypal' and not v:

raise ValueError('PayPal支付需要邮箱')

return v

四、嵌套模型

4.1 基础嵌套

from pydantic import BaseModel

class Address(BaseModel):

street: str

city: str

zip_code: str

country: str = "China"

class User(BaseModel):

id: int

name: str

address: Address

user = User(

id=1,

name="Alice",

address={"street": "123 Main St", "city": "Beijing", "zip_code": "100000"}

)

print(user.address.city) # Beijing

4.2 递归模型

from pydantic import BaseModel

from typing import Optional, List

class TreeNode(BaseModel):

value: int

children: Optional[List['TreeNode']] = None

使用字符串前向引用

TreeNode.model_rebuild()

tree = TreeNode(

value=1,

children=[

TreeNode(value=2),

TreeNode(value=3, children=[TreeNode(value=4)])

]

)

4.3 模型继承

from pydantic import BaseModel

class UserBase(BaseModel):

username: str

email: str

class UserCreate(UserBase):

password: str

class User(UserBase):

id: int

is_active: bool = True

class UserOut(User):

# 排除字段

model_config = {'exclude': {'password'}}

五、序列化

5.1 基础序列化

from pydantic import BaseModel

class User(BaseModel):

id: int

username: str

email: str

user = User(id=1, username="alice", email="alice@example.com")

转为字典

data = user.model_dump()

转为JSON

json_str = user.model_dump_json()

自定义序列化

data = user.model_dump(by_alias=True)

5.2 序列化配置

from pydantic import BaseModel, Field, ConfigDict

from datetime import datetime

class Event(BaseModel):

model_config = ConfigDict(

from_attributes=True,

populate_by_name=True,

str_strip_whitespace=True

)

id: int

name: str

start_time: datetime

# 自定义序列化

def model_dump(self, **kwargs):

dump = super().model_dump(**kwargs)

dump['start_time'] = self.start_time.isoformat()

return dump

5.3 排除字段

from pydantic import BaseModel, Field

class User(BaseModel):

id: int

username: str

password: str = Field(..., exclude=True) # 总是排除

email: str

user = User(id=1, username="alice", password="secret", email="a@e.com")

排除特定字段

data = user.model_dump(exclude={'password'})

data = user.model_dump(exclude={'id', 'password'})

条件排除

data = user.model_dump(exclude_unset=True) # 排除未设置的字段

六、数据转换

6.1 类型转换

from pydantic import BaseModel

class Config(BaseModel):

port: int # 自动转换

timeout: float

enabled: bool

config = Config(port="8080", timeout="30.5", enabled="true")

print(config.port) # 8080 (int)

print(config.timeout) # 30.5 (float)

print(config.enabled) # True (bool)

6.2 自定义类型

from pydantic import BaseModel, field_validator

from datetime import datetime

class Timestamp(BaseModel):

created_at: datetime

@field_validator('created_at', mode='before')

@classmethod

def parse_timestamp(cls, v) -> datetime:

if isinstance(v, str):

return datetime.fromisoformat(v)

if isinstance(v, (int, float)):

return datetime.fromtimestamp(v)

return v

使用

ts = Timestamp(created_at="2026-05-26T10:00:00")

ts2 = Timestamp(created_at=1716688800)

6.3 数据清洗

from pydantic import BaseModel, field_validator

class User(BaseModel):

username: str

email: str

@field_validator('username', 'email', mode='before')

@classmethod

def strip_whitespace(cls, v: str) -> str:

return v.strip() if isinstance(v, str) else v

@field_validator('email')

@classmethod

def lowercase_email(cls, v: str) -> str:

return v.lower()

user = User(username=" Alice ", email=" ALICE@EXAMPLE.COM ")

print(user.username) # Alice

print(user.email) # alice@example.com

七、FastAPI集成

7.1 请求体验证

from fastapi import FastAPI

from pydantic import BaseModel, Field

app = FastAPI()

class ItemCreate(BaseModel):

name: str = Field(..., min_length=1, max_length=50)

price: float = Field(..., gt=0)

description: str | None = Field(None, max_length=500)

tags: list[str] = Field(default_factory=list)

@app.post("/items/")

async def create_item(item: ItemCreate):

return item

7.2 响应模型

from fastapi import FastAPI

from pydantic import BaseModel, SecretStr

class UserIn(BaseModel):

username: str

password: SecretStr

email: str

class UserOut(BaseModel):

username: str

email: str

@app.post("/users/", response_model=UserOut)

async def create_user(user: UserIn):

# password不会被返回

return user

7.3 查询参数验证

from fastapi import FastAPI, Query

from typing import Annotated

app = FastAPI()

@app.get("/items/")

async def read_items(

q: Annotated[str | None, Query(min_length=3)] = None,

skip: Annotated[int, Query(ge=0)] = 0,

limit: Annotated[int, Query(ge=1, le=100)] = 100

):

return {"q": q, "skip": skip, "limit": limit}

八、性能优化

8.1 模型缓存

from pydantic import BaseModel

import time

class Data(BaseModel):

id: int

name: str

value: float

验证耗时

start = time.time()

for _ in range(10000):

Data(id=1, name="test", value=1.5)

print(f"v2耗时: {time.time() - start:.3f}s")

v1对比(如果安装了)

from pydantic.v1 import BaseModel as BaseModelV1

class DataV1(BaseModelV1):

id: int

name: str

value: float

8.2 序列化优化

from pydantic import BaseModel

import json

class LargeData(BaseModel):

id: int

name: str

items: list[dict]

metadata: dict

data = LargeData(

id=1,

name="test",

items=[{"id": i, "name": f"item{i}"} for i in range(100)],

metadata={"key": "value"}

)

快速序列化

json_str = data.model_dump_json()

排除大字段

json_str = data.model_dump_json(exclude={'items'})

九、实战案例

9.1 API请求/响应模型

from pydantic import BaseModel, Field, EmailStr

from typing import Optional

from datetime import datetime

class UserBase(BaseModel):

email: EmailStr

username: str = Field(..., min_length=3, max_length=50)

class UserCreate(UserBase):

password: str = Field(..., min_length=8)

class UserUpdate(BaseModel):

email: Optional[EmailStr] = None

username: Optional[str] = Field(None, min_length=3, max_length=50)

is_active: Optional[bool] = None

class User(UserBase):

id: int

is_active: bool = True

created_at: datetime

model_config = {'from_attributes': True}

class ItemBase(BaseModel):

name: str

price: float = Field(..., gt=0)

description: Optional[str] = None

class ItemCreate(ItemBase):

pass

class Item(ItemBase):

id: int

owner_id: int

created_at: datetime

model_config = {'from_attributes': True}

9.2 分页响应

from pydantic import BaseModel

from typing import Generic, TypeVar, List

T = TypeVar('T')

class Pagination(BaseModel):

page: int

page_size: int

total: int

total_pages: int

class PaginatedResponse(BaseModel, Generic[T]):

data: List[T]

pagination: Pagination

@classmethod

def from_list(cls, data: list[T], page: int, page_size: int, total: int):

total_pages = (total + page_size - 1) // page_size

return cls(

data=data,

pagination=Pagination(

page=page,

page_size=page_size,

total=total,

total_pages=total_pages

)

)

使用

response = PaginatedResponse.from_list(

data=[{"id": 1}, {"id": 2}],

page=1,

page_size=10,

total=25

)

常见坑自查清单

坑现象自查方法修复方案
v1/v2混用导入错误检查import统一用v2
ConfigDict缺失属性错误检查model_config用ConfigDict
前向引用NameError检查递归模型用字符串或rebuild
类型不匹配验证错误检查类型注解添加正确类型

结语

关键洞察

:

  • ●v2性能提升10倍
  • ●用model_config替代Config
  • ●支持前向引用和递归模型
  • ●FastAPI深度集成

互动

  1. 1.你升级到Pydantic v2了吗?
  2. 2.v2的性能提升明显吗?
  3. 3.用过哪些高级验证特性?
版本: V1.0 | 2026-05-26 | Python Web开发系列

📚 推荐阅读

📝 摘要:今天深入学习静态代码分析技术,这是安全审计的核心技能。从 Python AST 模块到检测模式设计,收获满满!

发布于 202603

01-Python 环境搭建与第一个脚本

发布于 202603

【优化】Python代码优化与调试技巧

发布于 202603

KEYWORDS

IL, Python, python, 字符串, 列表

💡 如果你觉得这篇文章有帮助,请点个在看,分享给更多需要的人!

📝 关注我,获取更多实用干货~

🤝 有问题欢迎评论区留言交流!

最新文章

随机文章