前言
Pydantic 是 Python 中最流行的数据验证与设置管理库,核心基于类型注解实现数据校验、序列化/反序列化,是 FastAPI 等框架的核心依赖。
核心概念与常用功能
环境准备
1
2
3
4
|
# 安装 Pydantic v2(推荐,性能比 v1 高)
pip install pydantic>=2.0
# 若需兼容 Python 3.8- 或旧代码,可装 v1(但建议迁移到 v2)
# pip install pydantic==1.10.14
|
数据模型定义与验证
Pydantic 的核心是 BaseModel,通过继承它定义数据模型,自动完成数据校验。
基础模型定义
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
|
from pydantic import BaseModel, Field
from typing import List, Optional, Dict
from datetime import datetime
# 定义基础模型
class User(BaseModel):
# 基础类型:必填字段
id: int
username: str
# 可选字段(None 表示允许为空)
email: Optional[str] = None
# 带默认值的字段(非必填)
age: int = 18
# 复杂类型:列表、字典
hobbies: List[str] = []
extra: Dict[str, str] = Field(default_factory=dict)
# 日期时间类型(自动解析字符串为 datetime 对象)
create_time: datetime = Field(default_factory=datetime.now)
# 正确数据:自动验证 + 类型转换
user1 = User(
id=1,
username="zhangsan",
email="zhangsan@example.com",
age=20,
hobbies=["reading", "coding"]
)
print(user1.id) # 1
print(user1.create_time) # 自动生成当前时间(datetime 对象)
print(user1.model_dump()) # 转为字典:{'id': 1, 'username': 'zhangsan', ...}
# 错误数据:触发 ValidationError
try:
# id 传入字符串(无法转为 int)、age 负数(后续会加校验)
User(id="not_int", username="lisi", age=-5)
except Exception as e:
print(e)
"""
输出校验错误:
1 validation error for User
id
Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='not_int', input_type=str]
age
Input should be greater than 0 [type=greater_than, input_value=-5, input_type=int]
"""
|
字段校验规则(Field 与校验器)
通过 Field 设置字段约束,或自定义校验器实现复杂规则:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
from pydantic import BaseModel, Field, field_validator, ValidationInfo
class User(BaseModel):
id: int = Field(ge=1, description="用户ID,必须≥1") # ge=大于等于
username: str = Field(min_length=3, max_length=20, description="用户名长度3-20")
age: int = Field(gt=0, lt=120, description="年龄0-120") # gt=大于,lt=小于
email: Optional[str] = Field(pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", description="邮箱格式")
# 自定义字段校验器(单个字段)
@field_validator("username")
def username_cannot_contain_space(cls, v):
if " " in v:
raise ValueError("用户名不能包含空格")
return v.strip() # 自动去除首尾空格
# 多字段关联校验(比如 email 必须包含 username)
@field_validator("email")
def email_contains_username(cls, v, info: ValidationInfo):
if v is None:
return v
username = info.data.get("username")
if username and username not in v:
raise ValueError(f"邮箱必须包含用户名 {username}")
return v
# 测试校验规则
user2 = User(
id=2,
username=" lisi ", # 自动去除空格
age=25,
email="lisi@example.com" # 包含 username "lisi",通过校验
)
print(user2.username) # lisi(已去空格)
# 错误案例:用户名含空格
try:
User(id=3, username="li si", age=30)
except Exception as e:
print(e) # 用户名不能包含空格
|
数据序列化/反序列化
Pydantic 提供便捷的方法实现模型与字典、JSON、字符串的转换:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
# 模型转字典
user = User(id=1, username="zhangsan", age=20)
print(user.model_dump()) # 完整字典
print(user.model_dump(exclude=["age"])) # 排除 age 字段
print(user.model_dump(include=["id", "username"])) # 只包含指定字段
# 模型转 JSON 字符串
print(user.model_dump_json(indent=2)) # 格式化输出
# {"id": 1, "username": "zhangsan", "age": 20, ...}
# 字典/JSON 转模型(反序列化)
user_dict = {"id": 2, "username": "lisi", "age": 25}
user3 = User(**user_dict) # 方法1:解包字典
user4 = User.model_validate(user_dict) # 方法2:官方推荐(更严谨)
# JSON 字符串转模型
user_json = '{"id": 3, "username": "wangwu", "age": 30}'
user5 = User.model_validate_json(user_json)
|
嵌套模型
支持模型嵌套,适用于复杂数据结构(如接口请求体):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
class Address(BaseModel):
province: str
city: str
detail: Optional[str] = None
class UserWithAddress(BaseModel):
id: int
username: str
address: Address # 嵌套 Address 模型
# 列表嵌套
addresses: List[Address] = []
# 嵌套模型使用
user = UserWithAddress(
id=1,
username="zhangsan",
address=Address(province="北京", city="北京市", detail="朝阳区"),
addresses=[
Address(province="上海", city="上海市"),
Address(province="广州", city="广州市")
]
)
print(user.address.city) # 北京市
print(user.addresses[0].province) # 上海
|
配置项(Model Config)
通过 model_config 自定义模型行为(v2 写法,v1 用 Config 类):
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
class User(BaseModel):
id: int
username: str
# 模型配置
model_config = {
"strict": False, # 非严格模式(默认):允许自动类型转换(如字符串转int)
"frozen": False, # 是否冻结模型(冻结后无法修改字段值)
"populate_by_name": True, # 支持按字段别名/名称赋值
"validate_assignment": True, # 修改字段值时重新校验
"json_schema_extra": { # 生成 OpenAPI 文档时的额外信息
"examples": [{"id": 1, "username": "zhangsan"}]
}
}
# 测试 validate_assignment
user = User(id=1, username="zhangsan")
user.age = -5 # 若模型有 age 字段,修改时会触发校验,报错
|
高级功能
自定义类型
针对通用校验规则,封装自定义类型,复用性更强:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
|
from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import CoreSchema, core_schema
from typing import Any
# 自定义手机号类型
class PhoneNumber(str):
@classmethod
def __get_pydantic_core_schema__(
cls, source_type: Any, handler: GetCoreSchemaHandler
) -> CoreSchema:
# 基础校验:字符串类型
core_schema = handler(str)
# 自定义校验规则:手机号格式
validate_schema = core_schema.no_info_plain_validator_function(
lambda v: cls.validate_phone(v)
)
# 组合校验规则
return core_schema.with_info_validator(validate_schema)
@classmethod
def validate_phone(cls, v: str) -> str:
import re
if not re.match(r"^1[3-9]\d{9}$", v):
raise ValueError("手机号格式错误(必须是13-19开头的11位数字)")
return v
# 使用自定义类型
class User(BaseModel):
name: str
phone: PhoneNumber
# 测试
user = User(name="zhangsan", phone="13812345678") # 正确
try:
User(name="lisi", phone="12345678901") # 错误格式
except Exception as e:
print(e) # 手机号格式错误
|
依赖注入与 FastAPI 集成
Pydantic 是 FastAPI 的核心,自动实现接口请求体校验、参数解析:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from fastapi import FastAPI
from pydantic import BaseModel, Field
app = FastAPI()
# 定义请求体模型
class ChatRequest(BaseModel):
query: str = Field(min_length=1, max_length=1000, description="用户查询")
user_id: Optional[str] = Field(None, description="用户ID")
stream: bool = Field(default=False, description="是否流式返回")
# 接口中使用模型,自动校验请求体
@app.post("/chat")
async def chat(request: ChatRequest):
return {
"code": 200,
"message": "success",
"data": {
"reply": f"你查询的是:{request.query}",
"stream": request.stream
}
}
|
- FastAPI 会自动基于 Pydantic 模型生成接口文档(OpenAPI);
- 请求体不符合规则时,自动返回结构化的错误信息。
数据设置管理(Settings)
Pydantic v2 内置 BaseSettings(需安装 pydantic-settings),用于管理环境变量/配置文件:
1
2
|
# 安装依赖
pip install pydantic-settings
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
from pydantic_settings import BaseSettings, SettingsConfigDict
from pathlib import Path
# 定义配置模型(从 .env 文件 + 环境变量读取)
class AppSettings(BaseSettings):
# 数据库配置
DB_HOST: str = "localhost"
DB_PORT: int = 3306
DB_USER: str = "root"
DB_PASS: str = "123456"
# 大模型配置
MODEL_NAME: str = "gpt-3.5-turbo"
MODEL_API_KEY: str = ""
# 配置读取规则
model_config = SettingsConfigDict(
env_file=".env", # 从 .env 文件读取
env_file_encoding="utf-8",
case_sensitive=False, # 不区分大小写(DB_HOST == db_host)
env_prefix="APP_", # 环境变量前缀(APP_DB_HOST 对应 DB_HOST)
)
# 初始化配置(自动加载 .env 文件 + 环境变量)
settings = AppSettings()
# 使用配置
print(settings.DB_HOST) # 从 .env 或环境变量读取
print(settings.MODEL_NAME)
|
.env 文件示例:
1
2
3
4
|
APP_DB_HOST=192.168.1.100
APP_DB_PORT=3306
APP_MODEL_NAME=gpt-4
APP_MODEL_API_KEY=sk-xxxxxx
|
批量数据校验
使用 pydantic.TypeAdapter 实现列表/批量数据的校验:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
from pydantic import BaseModel, TypeAdapter
class User(BaseModel):
id: int
username: str
# 校验单个用户
user_adapter = TypeAdapter(User)
user = user_adapter.validate_python({"id": 1, "username": "zhangsan"})
# 校验用户列表(批量)
users_adapter = TypeAdapter(list[User])
users = users_adapter.validate_python([
{"id": 1, "username": "zhangsan"},
{"id": 2, "username": "lisi"}
])
print(users) # [User(id=1, username='zhangsan'), User(id=2, username='lisi')]
|
性能优化技巧
Pydantic v2 已大幅提升性能,结合以下技巧进一步优化:
- 使用
model_validate 而非手动解包:User.model_validate(data) 比 User(** data) 更快;
- 关闭不必要的校验:
model_config = {"validate_assignment": False}(修改字段时不校验);
- 使用严格模式:
model_config = {"strict": True}(禁用自动类型转换,提升性能);
- 批量校验用 TypeAdapter:比循环创建模型更高效;
- 缓存模型 schema:复杂模型可缓存
model_json_schema() 结果,避免重复生成。
错误处理与自定义异常
自定义校验错误信息,提升可读性:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
from pydantic import BaseModel, field_validator, ValidationError
from pydantic_core import ErrorDetails
class User(BaseModel):
id: int
username: str
@field_validator("id")
def id_must_be_positive(cls, v):
if v <= 0:
# 自定义错误信息和类型
raise ValueError(f"用户ID {v} 必须大于0", code="id_negative")
return v
# 捕获异常并解析错误信息
try:
User(id=-1, username="zhangsan")
except ValidationError as e:
# 解析错误详情
errors: list[ErrorDetails] = e.errors()
for err in errors:
print(f"字段:{err['loc'][0]}, 错误类型:{err['type']}, 信息:{err['msg']}")
# 输出:字段:id, 错误类型:value_error, 信息:用户ID -1 必须大于0
|
总结
基础核心
BaseModel:Pydantic 的核心,定义数据模型并自动校验;
Field:设置字段约束(长度、范围、正则等);
field_validator:自定义字段校验规则(单字段/多字段关联);
- 序列化/反序列化:
model_dump()/model_validate() 实现模型与字典/JSON 的转换。
进阶核心
- 自定义类型:封装通用校验规则,提升复用性;
- Settings 管理:统一管理环境变量/配置文件;
- TypeAdapter:批量数据校验的高效方式;
- FastAPI 集成:自动实现接口请求体校验和文档生成;
- 性能优化:严格模式、关闭不必要校验、批量处理。