跳到主要内容

Pydantic 模型

前言

Pydantic 是一个使用 Python 类型注解进行数据验证和管理的库,可在运行时强制类型校验。FastAPI 深度整合了 Pydantic:请求体自动校验、响应自动序列化、OpenAPI 文档自动生成。

本笔记基于 Pydantic 2.x 整理。v1 与 v2 的主要差异:

  • dict() / json()model_dump() / model_dump_json()
  • validatorfield_validator(配合 @classmethod
  • class Configmodel_config = ConfigDict(...)schema_extrajson_schema_extra
  • Optional[str] 推荐写成 str | None
备注

EmailStrNameEmail 等邮箱类型依赖 email-validator,使用前需安装:python -m pip install "pydantic[email]"

模型常见类型

from datetime import date
import json

from pydantic import (
BaseModel,
DirectoryPath,
EmailStr,
Field,
FilePath,
HttpUrl,
IPvAnyAddress,
Json,
NameEmail,
SecretBytes,
SecretStr,
)


class Tmp1(BaseModel):
name: str # 字符串类型
age: int # 整型
age2: int | None # 可选参数,赋值则必须是整型
age3: int | None = None # 可选参数,默认值为 None
enable: bool
hobby: list
address: dict
birthday: date

email: EmailStr # 电子邮件格式
email2: NameEmail # 带用户名的电子邮件格式
filepath: FilePath # 已存在的文件路径
directorypath: DirectoryPath # 已存在的目录路径
ipaddr: IPvAnyAddress # IP 地址
secretstr: SecretStr # 敏感字符串,repr/输出时自动打码
secretbyte: SecretBytes # 敏感二进制
website: HttpUrl # URL 地址
jsonobj: Json # JSON 字符串,自动解析为 dict/list


if __name__ == "__main__":
v1 = Tmp1(
name="zhangsan",
age=18,
age2=19,
enable=True,
hobby=["sing", "dance", "basketball"],
address={
"Country": "China",
"Province": "Anhui",
},
birthday="2023-01-01",
email="zhangsan@lisi.com",
email2="zhangsan <zhangsan@lisi.com>",
filepath="./test.txt",
directorypath="./test",
ipaddr="127.0.0.1",
secretstr="123456qwerty",
secretbyte=b"qwerty123456",
website="https://baidu.com",
jsonobj=json.dumps({"hello": "world"}),
)
print(v1)
print("-" * 50)
print(v1.model_dump()) # 输出 dict
print("-" * 50)
print(v1.model_dump_json()) # 输出 JSON 字符串

运行输出:

name='zhangsan' age=18 age2=19 age3=None enable=True hobby=['sing', 'dance', 'basketball'] address={'Country': 'China', 'Province': 'Anhui'} birthday=datetime.date(2023, 1, 1) email='zhangsan@lisi.com' email2=NameEmail(name='zhangsan', email='zhangsan@lisi.com') filepath=PosixPath('test.txt') directorypath=PosixPath('test') ipaddr=IPv4Address('127.0.0.1') secretstr=SecretStr('**********') secretbyte=SecretBytes(b'**********') website=Url('https://baidu.com/') jsonobj={'hello': 'world'}
--------------------------------------------------
{'name': 'zhangsan', 'age': 18, 'age2': 19, 'age3': None, 'enable': True, 'hobby': ['sing', 'dance', 'basketball'], 'address': {'Country': 'China', 'Province': 'Anhui'}, 'birthday': datetime.date(2023, 1, 1), 'email': 'zhangsan@lisi.com', 'email2': 'zhangsan <zhangsan@lisi.com>', 'filepath': 'test.txt', 'directorypath': 'test', 'ipaddr': '127.0.0.1', 'secretstr': '**********', 'secretbyte': '**********', 'website': 'https://baidu.com/', 'jsonobj': {'hello': 'world'}}
--------------------------------------------------
{"name":"zhangsan","age":18,"age2":19,"age3":null,"enable":true,"hobby":["sing","dance","basketball"],"address":{"Country":"China","Province":"Anhui"},"birthday":"2023-01-01","email":"zhangsan@lisi.com","email2":"zhangsan <zhangsan@lisi.com>","filepath":"test.txt","directorypath":"test","ipaddr":"127.0.0.1","secretstr":"**********","secretbyte":"**********","website":"https://baidu.com/","jsonobj":{"hello":"world"}}
提示

SecretStr / SecretBytesrepr() 和序列化时会打码,适合密码、Token 等敏感字段。

Field 字段声明与校验

from pydantic import BaseModel, Field, ValidationError


class Tmp2(BaseModel):
name: str = Field(..., title="姓名", description="用户姓名", max_length=10, examples=["张三"])
age: int = Field(..., title="年龄", description="用户年龄", gt=0, lt=150, examples=[18])


if __name__ == "__main__":
try:
v2 = Tmp2(name="张三", age=188)
except ValidationError as e:
print(e.errors())
else:
print(v2.model_dump())

运行输出:

[{'type': 'less_than', 'loc': ('age',), 'msg': 'Input should be less than 150', 'input': 188, 'ctx': {'lt': 150}, 'url': 'https://errors.pydantic.dev/2.5/v/less_than'}]

自定义校验器 field_validator

from pydantic import BaseModel, ValidationError, field_validator


class Tmp3(BaseModel):
name: str
age: int

@field_validator("age")
@classmethod
def valid_age(cls, v: int) -> int:
if v < 0:
raise ValueError("年龄必须大于0")
if v > 150:
raise ValueError("年龄必须小于150")
return v


if __name__ == "__main__":
try:
v2 = Tmp3(name="张三", age=188)
except ValidationError as e:
print(e.errors())
else:
print(v2.model_dump())
备注

旧版本写法为 @validator("age"),v2 已废弃,统一使用 @field_validator(...) + @classmethod。校验器里 raise ValueError(...) 会被转换为字段校验错误。

嵌套模型

模型可以嵌套引用其它模型,用于复杂请求/响应结构:

from datetime import date

from pydantic import BaseModel, ConfigDict, Field


class CityInfo(BaseModel):
name: str = Field(..., examples=["Beijing"])
country: str
country_code: str | None = None
country_population: int = Field(
default=800, ge=800, title="人口数量", description="国家的人口数量"
)

# v2 写法:给 OpenAPI 提供示例
model_config = ConfigDict(
json_schema_extra={
"example": {
"name": "Shanghai",
"country": "China",
"country_code": "CN",
"country_population": 1400000000,
}
}
)


class Data(BaseModel):
city: list[CityInfo] | None = None
date: date
confirmed: int = Field(default=0, ge=0, description="确诊数")
death: int = Field(default=0, ge=0, description="死亡数")
recovered: int = Field(default=0, ge=0, description="痊愈数")


@app.put("/request_body/nested")
def nested_models(data: Data):
return data
备注

v1 的 class Config: schema_extra = {...} 已废弃,v2 使用 model_config = ConfigDict(json_schema_extra={...})

复杂类型:List / Set / Dict

from pydantic import BaseModel


class ComplexModel(BaseModel):
tags: list[str] = [] # 字符串列表
scores: dict[str, int] = {} # 键为字符串、值为整数的字典
unique_names: set[str] = set() # 字符串集合(注意默认值要用 set(),不能用 [])
matrix: list[list[int]] = [] # 二维列表


m = ComplexModel(tags=["a", "b"], scores={"math": 90}, unique_names={"x", "y"})
print(m.model_dump())