Python Pydantic库使用教程:从入门到精通
文章目录
Pydantic 完全指南:从入门到实战
一、Pydantic 简介
1.1 什么是 Pydantic?
Pydantic 是一个用于数据验证和设置管理的 Python 库,它通过 Python 类型注解来定义数据模型,提供强大的数据验证功能。
1.2 核心功能
1.3 主要特性
1.4 Pydantic 生态系统
FastAPI 深度集成 Pydantic,用于请求/响应模型验证和自动文档生成。
相关工具
二、安装
版本不同,方法差异较大。固定该版本来使用
pip install pydantic
三、基础使用
3.1 基本数据模型定义
from pydantic import BaseModel
class User(BaseModel):
id: int
name: str
3.2 数据验证示例
user = User(id=1, name="John")
print(user) # 输出验证后的模型
try:
invalid_user = User(id="one", name=123)
except ValidationError as e:
print(e) # 输出验证错误
四、高级功能
4.1 自定义验证器
4.1.1 基本验证器
from pydantic import BaseModel, field_validator
class User(BaseModel):
name: str
age: int
@field_validator('name')
@classmethod
def name_must_contain_space(cls, v):
if ' ' not in v:
raise ValueError('必须包含空格')
return v.title()
4.1.2 多字段验证
@field_validator('age')
@classmethod
def age_must_be_positive(cls, v):
if v <= 0:
raise ValueError('必须是正整数')
return v
4.2 嵌套模型
class Address(BaseModel):
street: str
city: str
class User(BaseModel):
name: str
address: Address
4.3 模型配置:定制化模型行为
class ConfigModel(BaseModel):
name: str
class Config:
allow_mutation = False # 使实例不可变
extra = 'forbid' # 禁止额外字段
4.3.1 常用配置选项详解
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optional
class SystemSettings(BaseModel):
"""系统配置模型"""
class Config:
# 基础配置
allow_mutation = False # 模型实例不可变
extra = 'forbid' # 禁止额外字段
validate_all = True # 验证所有字段(即使没有提供值)
# 日期时间处理
json_encoders = {
datetime: lambda v: v.strftime('%Y-%m-%d %H:%M:%S')
}
# 错误信息模板
error_msg_templates = {
'value_error.missing': '字段 {field} 是必填项',
'type_error.integer': '字段 {field} 必须是整数'
}
hostname: str = Field(..., min_length=3, max_length=63)
port: int = Field(1024, ge=1024, le=65535)
last_updated: Optional[datetime] = None
debug_mode: bool = False
# 使用示例
try:
settings = SystemSettings(
hostname="my-server",
port=8080, # 端口不在允许范围内
extra_field="value" # 额外字段会触发错误
)
except ValidationError as e:
print(e.json(indent=2))
4.3.2 高级配置技巧
from pydantic import BaseModel, validator
import os
class AppConfig(BaseModel):
"""应用配置模型,支持从环境变量加载"""
class Config:
env_prefix = 'APP_' # 环境变量前缀
case_sensitive = False # 不区分大小写
allow_population_by_field_name = True # 允许通过字段名填充
db_host: str = "localhost"
db_port: int = 5432
db_user: str
db_pass: str
@validator('db_pass')
def validate_password(cls, value):
if len(value) < 8:
raise ValueError('数据库密码至少需要8个字符')
return value
# 设置环境变量
os.environ['APP_DB_USER'] = 'admin'
os.environ['APP_DB_PASS'] = 'securepassword123'
# 从环境变量创建配置
config = AppConfig()
print("数据库配置:", config.dict())
4.3.3 自定义JSON序列化
from pydantic import BaseModel
from decimal import Decimal
from datetime import date
class FinancialRecord(BaseModel):
transaction_id: str
amount: Decimal
date: date
description: str = None
class Config:
json_encoders = {
Decimal: lambda v: str(round(v, 2)),
date: lambda v: v.isoformat()
}
# 控制dict输出
@staticmethod
def schema_extra(schema, model):
schema['example'] = {
"transaction_id": "txn_12345",
"amount": "100.00",
"date": "2023-01-01",
"description": "Sample transaction"
}
record = FinancialRecord(
transaction_id="txn_98765",
amount=Decimal("1234.5678"),
date=date.today()
)
print("JSON输出:", record.json(indent=2))
print("字典输出:", record.dict())
五、实际应用场景
5.1 Web 开发(FastAPI)
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@app.post("/items/")
async def create_item(item: Item):
return item
5.2 API 开发
class UserResponse(BaseModel):
id: int
name: str
email: str
def get_user() -> UserResponse:
return UserResponse(id=1, name="Alice", email="alice@example.com")
5.3 数据处理
def process_data(raw_data: dict) -> User:
try:
return User(**raw_data)
except ValidationError as e:
print(f"数据验证失败: {e}")
raise
六、方法与配置全解
6.1 模型方法大全
user = User(id=1, name="John")
print(user.model_dump()) # 输出: {'id': 1, 'name': 'John'}
print(User.model_json_schema()) # 输出模型的JSON Schema
6.1.1 基础模型方法
from pydantic import BaseModel
from datetime import datetime
class User(BaseModel):
id: int
name: str
created_at: datetime = None
# 1. 构造方法
user = User(id=1, name="Alice") # 标准构造
user = User.parse_obj({"id": 2, "name": "Bob"}) # 从字典构造
# 2. 序列化方法
print(user.dict()) # 转为字典
# 输出: {'id': 1, 'name': 'Alice', 'created_at': None}
print(user.json()) # 转为JSON字符串
# 输出: {"id": 1, "name": "Alice", "created_at": null}
# 3. 拷贝方法
user_copy = user.copy() # 浅拷贝
user_deepcopy = user.copy(deep=True) # 深拷贝
# 4. 字段访问方法
print(user.dict(exclude={'created_at'})) # 排除特定字段
print(user.dict(include={'id'})) # 只包含特定字段
6.1.2 高级模型方法
# 1. 模式导出
print(User.schema()) # 获取JSON Schema
print(User.schema_json(indent=2)) # 获取格式化的JSON Schema
# 2. 字段信息获取
print(User.__fields__) # 获取所有字段定义
print(User.__fields__['id'].type_) # 获取字段类型(int)
# 3. 更新方法
updated_user = user.copy(update={"name": "Alice Smith"})
# 4. 解析原始数据
raw_data = '{"id": 3, "name": "Carol"}'
user = User.parse_raw(raw_data) # 从JSON字符串解析
6.2 模型配置全解
6.2.1 基础配置选项
class ConfigDemo(BaseModel):
name: str
age: int
class Config:
# 字段处理配置
allow_mutation = False # 禁止修改模型实例
extra = 'forbid' # 处理额外字段: 'allow'|'ignore'|'forbid'
underscore_attrs_are_private = True # 下划线开头属性为私有
# 验证配置
validate_all = True # 验证所有字段(包括可选字段)
validate_assignment = True # 赋值时验证
# 序列化配置
json_encoders = {
datetime: lambda v: v.isoformat() # 自定义JSON编码
}
json_loads = lambda x: x # 自定义JSON解析
json_dumps = lambda x, **kwargs: x # 自定义JSON序列化
6.2.2 环境变量配置
import os
# 在我使用的v2版本中已经弃用
# from pydantic import BaseSettings
from pydantic_settings import BaseSettings
class AppSettings(BaseSettings):
# 自动从环境变量读取(不区分大小写)
db_host: str = 'localhost'
db_port: int = 5432
class Config:
env_prefix = 'APP_' # 环境变量前缀(APP_DB_HOST)
case_sensitive = False # 不区分大小写
env_file = '.env' # 从.env文件加载
env_file_encoding = 'utf-8'
os.environ['APP_DB_HOST'] = 'prod.db.example.com'
settings = AppSettings()
# 在我使用的v2版本中已经弃用
# print(settings.dict())
print(settings.model_dump())
输出:
{'db_host': 'prod.db.example.com', 'db_port': 5432}
6.2.3 高级配置技巧
from typing import Any
from pydantic import BaseModel
from pydantic import Field
class AdvancedConfig(BaseModel):
value: Any = Field(
alias='VALUE',
title='配置值',
description='可以是任何类型的值'
)
class Config:
# 自定义schema额外信息
@staticmethod
def json_schema_extra(schema: dict, model) -> None:
schema['examples'] = [{'VALUE': 42}]
# 使用别名创建实例
config = AdvancedConfig.model_validate({'VALUE': 'example'})
print(config.model_dump_json(indent=2))
输出:
{
"value": "example"
}
6.3 字段配置详解
6.3.1 Field 函数参数
import json
from pydantic import BaseModel, Field
from typing import Optional
class Product(BaseModel):
# 基础字段配置
id: int = Field(..., gt=0, description="产品ID必须为正整数")
name: str = Field(
...,
min_length=2,
max_length=100,
pattern=r'^[a-zA-Z0-9_\- ]+$',
title="产品名称",
description="2-100个字符,允许字母、数字、空格、下划线和连字符"
)
# 可选字段配置
price: Optional[float] = Field(
None,
gt=0,
alias="productPrice",
example=99.99
)
# 复杂配置
tags: list[str] = Field(
default_factory=list,
min_items=1,
max_items=10,
description="产品标签列表"
)
# 使用示例
product = Product(
id=1,
name="Premium Product",
price=199.99,
tags=["new", "featured"]
)
print(json.dumps(product.model_json_schema(), indent=2))
6.3.2 自定义字段验证
from pydantic import validator
class CustomFieldDemo(BaseModel):
username: str
password: str = Field(..., min_length=8)
@validator('username')
def username_alphanumeric(cls, v):
if not v.isalnum():
raise ValueError('必须是字母数字组合')
return v
@validator('password')
def password_complexity(cls, v):
if not any(c.isupper() for c in v):
raise ValueError('必须包含大写字母')
if not any(c.isdigit() for c in v):
raise ValueError('必须包含数字')
return v
class Config:
fields = {
'password': {'exclude': True} # 从dict()/json()输出中排除
}
# 使用示例
user = CustomFieldDemo(username="Alice123", password="Pass1234")
print(user.dict()) # 密码字段被排除
6.4 实用方法组合示例
6.4.1 模型更新模式
class UpdateModel(BaseModel):
name: str = None
age: int = None
class Config:
extra = 'forbid'
validate_assignment = True
def apply_update(self, **kwargs):
"""安全更新方法"""
for field, value in kwargs.items():
if field in self.__fields__:
setattr(self, field, value)
user = UpdateModel()
user.apply_update(name="Alice", age=30, invalid="value") # 忽略无效字段
print(user.dict())
6.4.2 动态模型创建
from pydantic import create_model
DynamicModel = create_model(
'DynamicModel',
id=(int, Field(..., gt=0)),
name=(str, Field(..., max_length=100)),
__config__=dict(extra='forbid')
)
instance = DynamicModel(id=1, name="Test")
print(instance)
6.4.3 模型继承与扩展
class BaseUser(BaseModel):
id: int
username: str
class AdminUser(BaseUser):
permissions: list[str]
is_active: bool = True
class Config:
allow_mutation = False
admin = AdminUser(
id=1,
username="admin",
permissions=["read", "write", "delete"]
)
print(admin.json(indent=2))
6.5 验证器使用技巧
6.5.1 验证器顺序问题
# 正确顺序
@field_validator('field')
@classmethod
def validator(cls, v):
return v
# 错误顺序(可能失效)
@classmethod
@field_validator('field')
def validator(cls, v):
return v
6.5.2 高级验证选项
@field_validator('field', mode='before') # 在基本验证前执行
@field_validator('field', check_fields=False) # 不检查字段是否存在
七、学习资源
7.1 官方文档
Pydantic 官方文档 是最全面的学习资源。
7.2 社区支持
作者:Hellchin