python之pydantic使用小结
文章目录
- 简介
- 前置相关知识点
- Annotated类型注解增强工具
- python内置 dataclasses.dataclass 装饰器
- generic type 泛型
- pydantic 用法汇总
- schema的四种校验方式
- 数据模型类
- model copy 模型实例拷贝
- model validate 基本的数据校验
- model dump 基本的模型序列化.
- generic model 泛型模型
- dynamic model 动态模型
- 模型字段与类型详解
- 模型的 json schema
- customized validator 定制化校验器
- 字段级别校验器
- 后置(After)校验器
- 前置(Before)校验器
- 普通(Plain)验器
- 包装(Wrap)校验器
- 自定义校验器异常的细节.
- ValidtionInfo可选参数
- 模型级别校验器
- 后置(After)校验器
- 前置(Before)校验器
- 包装(Wrap)校验器
- customized serializer 定制序列化
- 字段级别序列化器
- 普通(Plain)序列化器
- 包装(Wrap)序列化器
- FieldSerializationInfo 细节部分
- 模型级别序列化器
- 普通(Plain)序列化器
- 包装(Wrap)序列化器
- pydantic.dataclasses.dataclass 修饰类定义schema
- TypeAdapter定义schema
- validate_call 校验
- 实战案例
- llm function schema
- llm 结构化输出以及解析
- 小结
简介
pydantic是一个python中使用非常广泛的数据校验lib,官方文档地址:https://docs.pydantic.dev/latest,本文详细总结这个lib的常规的使用方式和技巧,非常规以及低频率使用的技巧本文不会涉及,需要使用到低频的用法请移步官方文档.
前置相关知识点
Annotated类型注解增强工具
先要聊一下python的类型注解,首先python是弱类型语言,它不像java那样在编写代码的时候必须强制指定类型,python解释器本身也不会在运行时进行类型检查,类型问题只会在运行时候报错,举下面一个例子.
def test_str(a):res = a.split(' ')print(res)return resif __name__ == '__main__':test_str(10)
本质上test_str函数本身是要接收一个字符串类型参数,但是实际传递了一个整数类型的参数,实际运行时就会报错.下面改一下代码,给参数加上类型注解:
def test_str(a: str):res = a.split(' ')print(res)return resif __name__ == '__main__':test_str(10)
这个时候如果静态分析工具开启了类型检查功能(笔者使用vscode开发,静态分析工具是pylance插件,插件本身是基于pyright的),这个时候在调用处test_str(10)就会高亮错误:
无法将“Literal[10]”类型的参数分配给函数“test_str”中类型为“str”的参数“a”
“Literal[10]”不可分配给“str”PylancereportArgumentType
类型检查提示存在类型问题,虽然这个时候强制运行也能运行且抛出异常(python解释器不会进行运行时类型检查,静态分析工具也不会阻止运行程序,只是语法高亮提示存在类型问题).但至少在运行之前,开发者就获知了代码潜在的类型问题,在代码运行之前就可以修复这些问题. 这就是类型注解带来的好处,配合静态工具中的类型检查在开发阶段就降低类型问题带来的风险.
vscode 使用 Pylance 静态分析工具做类型检查,但是出于性能考虑,此类型检查功能默认是关闭,相关配置是python.analysis.typeCheckingMode默认是off
python.analysis.typeCheckingMode
Used to specify the level of type checking analysis performed.
Default: off.
Note that the value of this setting can be overridden by having a pyrightconfig.json or a pyproject.toml. For more information see this link.
Available values:
off: No type checking analysis is conducted; unresolved imports/variables diagnostics are produced.
basic: All rules from off + basic type checking rules.
standard: All rules from basic + standard type checking rules.
strict: All rules from standard + strict type checking rules.
You can refer to pyright documentation to reference the default type checking rules for each of the type checking modes.
Performance Consideration:
Setting python.analysis.typeCheckingMode to off can improve performance by disabling type checking analysis, which can be resource-intensive, especially in large codebases.
这里需要配置开启,由于这个功能会影响性能,所以一般不选择全局开启,它支持在项目级别开启,比如项目中的pyrightconfig.json或者是pyproject.toml中配置. 由于现代的python项目基本都是基于uv构建的,所以项目中自然存在pyproject.toml文件,因此在此文件中配置开启即可,如下是一个最简单的配置,将checkmode设置为standard.
[project]
name = "test"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["pydantic>=2.11.9",
][tool.pyright]
typeCheckingMode = "standard"
pythonVersion = "3.11"
聊完python基本的类型注解之后,回到Annotated这个工具类. Annotated本质上是类型注解工具类,可将一个或者多个元数据添加给指定类型,增强类型功能.可配合静态分析工具或某些框架(pydantic)使用以实现更丰富的功能(pydantic数据校验).具体用法是Annotated[T, meta1, meta2,...]即必须有指定的具体类型T,和至少1个元数据.元数据可以是任何python类型对象(字符串,类,类实例,函数等)它可以用于表达对类型取值的各种约束.元数据就是提供给静态分析工具或者框架解析使用的, python解释器本身不会去处理Annotated中的元数据,因此即便定义了约束,python解释器运行时本身也不处理这些约束且目前的静态分析工具类型检查功能也不会去处理这些元数据的约束条件.
举个简单的例子.
from annotated_types import Le, Ge
# meta 约束参数a取值范围是10到20,但是实际传递一个不满足约束条件的2也能运行
# 类型检查也不会去处理meta表示约束条件,因此这里不会有高亮的错误提示
def test_meta(a: Annotated[int, Ge(10), Le(20)]):print(a)if __name__ == '__main__':test_meta(2)
结果是能正常运行,且类型检查不会高亮错误.
结合pydantic举一个简单的例子.
from pydantic import BaseModel, ValidationError
from annotated_types import Le, Ge, MinLen, MaxLenclass A(BaseModel, strict=True):# name长度不小于5name: Annotated[str, MinLen(5)]# age 在0到100age: Annotated[int, Ge(0), Le(100)]if __name__ == '__main__':try:a = A(name='a', age=101)except ValidationError as e:print(e)
运行结果:
2 validation errors for A
nameString should have at least 5 characters [type=string_too_short, input_value='a', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/string_too_short
ageInput should be less than or equal to 100 [type=less_than_equal, input_value=101, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/less_than_equal
可见Annotated中携带的元数据约束信息,pydantic可以正确的识别并且正确处理.这也是Annotated这个类型注解工具类在pydantic中非常重要的原因.
关于Annotated这个类型注解工具类还涉及到三个重要的工具方法,分别是get_origin,get_args以及get_type_hints,下面分别讲解这三个方法的作用.
t: TypeAlias = Annotated[int, Field(le=10)]
def test_arg_func(a: Annotated[str, MaxLen(10)]) -> str:return adef three_tools():# get_origin, get_args, get_type_hints# get_origin 获取无下标的类型,比如有类型参数的类型X[Y,Z, ...] 返回Xres = get_origin(t)# 这里返回Annotatedprint(res)res = get_origin(str)# 返回Noneprint(res)# get_args 返回带参数的类型参数列表,返回值是一个元祖res = get_args(t)# 这里返回(str, FieldInfo)print(res)res = get_args(str)# 这里是空元祖print(res)# get_type_hints 返回一个对象的类型注解 include_extras 为True 会返回完整的类型信息,比如参数使用Annotated注解# 返回结果是一个字典,入参可是一个模型或者函数res = get_type_hints(test_arg_func, include_extras=True)# 返回形参参数名和其完整类型,返回类型也会出现在结果字典里面print(res)if __name__ == '__main__':three_tools()
输出:
<class 'typing.Annotated'>
None
(<class 'int'>, FieldInfo(annotation=NoneType, required=True, metadata=[Le(le=10)]))
()
{'a': typing.Annotated[str, MaxLen(max_length=10)], 'return': <class 'str'>}
python内置 dataclasses.dataclass 装饰器
dataclass装饰器主要用于装饰数据装载类,根据类中声明的带类型注解的类属性,自动生成__init__,__repr__,__eq__等(默认生成这三个,可通过设置装饰器实现更多魔术函数生成)魔术函数.注意使用时生成的__init__方法参数和类中定义的有效属性按照声明顺序一一对应,这样初始化的实例就具有这些类属性同名的实例属性.注意这里的类属性一定需要带有类型注解,没有类型注解的属性不会出现在__init__方法参数列表里面,进而初始化的实例也不存在这个实例属性.
举个例子:
@dataclass
class B:# 带类型声明的类属性才会被视作字段,出现在__init__初始化参数列表里面name: str = 'abc'# age 没有类型注解的属性不会出现在__init__初始化参数列表里面age = 10if __name__ == '__main__':# 这里__init__只有name没有age,实例本身也不存在age这个实例属性b = B('bcd')# 打印__dict__中实例属性只有name,没有age.print(b.__dict__)
输出
{'name': 'bcd'}
正常定义的例子:
@dataclass
class D:# 注意没有默认值的类属性定义要在有默认值的前面age: int# 默认值也会出现在__init__方法中作为参数默认值,所以在定义时无默认值的属性要在有默认值属性之前name: str = 'bcd'if __name__ == '__main__':d = D(age=5)print(d.__dict__)
输出
'age': 5, 'name': 'bcd'}
后面pydantic lib中会涉及到一个同名的装饰器.
generic type 泛型
pydantic 会涉及到部分python泛型的知识点,所以这里先回顾一下python中的泛型,完整泛型的介绍移步官方文档: https://typing.python.org/en/latest/reference/generics.html.
from typing import Generic, TypeVar# 定义类型变量的TypeVar类,表示可以被不同类型替换的占位符
T = TypeVar('T')# 标准泛型类用法,继承自Generic[T],Generic 是一个基类,用于声明类是"泛型类",即它接受类型参数
class Stack(Generic[T]):def __init__(self):self.items: list[T] = []def push(self, item: T):self.items.append(item)def pop(self) -> T:return self.items.pop()def empty(self):return not self.itemsdef test():# 构建实例的时候没有强制指定类型参数s1 = Stack()s1.push('abc')# 创建实例的具体类型是int,但是实际可以传递非int数据,代码检查工具会高亮错误但是解释器不会阻止运行s2 = Stack[int]()s2.push('abc')s2.push(1)
泛型类型推断,类型参数出现在初始化函数时,类型可以被推断,不用显式指定类型.
# 泛型类型参数推断问题
class Box(Generic[T]):# 初始化函数有类型参数def __init__(self, item: T):self.item = itemdef print(self):print(type(self.item))def test():# 实例化可不指定类型参数具体类型,根据初始化方法传递参数推断类型b = Box('123')b.print()# 如果显示指定类型参数具体类型一定需要和初始化函数传参类型一致# 不一样类型检查工具会报错,实际能正常运行,且实际类型是初始化函数传递的参数的类型b1 = Box[int]('123')b1.print()
泛型函数和泛型方法:
首先是泛型函数,它不属于任何类,属于模块中的函数.多个泛型函数可共用同一个类型变量,函数调用时类型互不影响.
# 泛型函数和泛型方法
T = TypeVar('T')# 作用域在函数内,多个函数可以共用一个类型变量.
# 实际调用时,函数的类型参数互相独立不会有任何关联.
def first(arg: Sequence[T]) -> T:return arg[0]def last(arg: Sequence[T]) -> T:return arg[-1]def test():print(first('1234'))print(last([1,2,3]))if __name__ == '__main__':test()
输出:
1
3
泛型方法: 主要是针对类中的方法,普通类中的方法参数或者返回值带类型变量则此方法属于泛型方法;泛型类中的方法如果带了和本类绑定的类型变量之外的类型变量则也被称为泛型方法.举个例子:
T = TypeVar('T')
U = TypeVar('U')class Normal:def __init__(self, arg: str):self.arg = argdef pair(self, arg: T) -> tuple[str, T]:return (self.arg, arg)class GenericDemo(Generic[T]):def __init__(self, arg: T):self.arg = argdef pair(self, arg: U) -> tuple[T, U]:return (self.arg, arg)def test():print(Normal('1').pair(2))print(GenericDemo[int](1).pair('2'))if __name__ == '__main__':test()
输出结果:
('1', 2)
(1, '2')
类型参数的约束问题:一般定义的类型参数无任何约束,理论上使用任何类型代码检查工具都不会报错,这样运行时就存在出错的风险.目前类型参数支持使用bound或者constraints给类型添加一些约束.bound表示类型的上界,传递具体类型则是要求类型参数具体取值必须是bound对应类型或者其子类,其次bound还支持协议类型(typing.SupportsXXX 等类型),要求类型参数具体取值类型必须满足协议类型中约定的方法以及返回值.举例如下:
bound边界为具体类型:
class TBound:def __init__(self, arg):self.arg = arg# 3.11版本 TypeVar 不支持default, 只有constraints和bound能约束类型范围
# 且bound和constarints 是互斥的,定义的时候只能二者选其一
T_bound = TypeVar('T_bound', bound=TBound)class BoundClass(Generic[T_bound]):def __init__(self, arg: T_bound):self.arg = argdef print(self):print(type(self.arg))def test():# 类型检查工具报错int不满足类型变量T_bound的上限类型 TBound#test = BoundClass[int](arg=123)test = BoundClass[TBound](arg=TBound(arg=123))test.print()if __name__ == '__main__':test()
运行结果:
<class '__main__.TBound'>
边界为协议类型:
from typing import SupportsAbs# 实现__abs__方法且返回类型为float
T_support = TypeVar('T_support', bound=SupportsAbs[float])def show_type(arg1: T_support, arg2: T_support):print(type(arg1), type(arg2))def test():# float类型满足协议类型类的要求show_type(1.2, 1.3)# str类型则不满足要求,代码检查工具会高亮类型不匹配的错误show_type('1', '2')
constraints表示类型参数必须为指定的约束类型集合中的其中一种,举例如下:
# 类型名后多个类型参数表示类型约束集合,实际取值只能在这些约束类型之中选择一个.
T_cons = TypeVar('T_cons', str, float)def show_all_types(arg1: T_cons, arg2: T_cons):print(type(arg1), type(arg2))def test():# str类型在约束类型集合中show_all_types('a', 'b')# float类型在约束类型集合中show_all_types(1.2, 2.3)# 注意实际类类型只能是约束类型集合中一个,而不是集合中所有类型的union 比如 str | float# 所以这里会高亮错误,一个类型是str另一个是float,这里必须要么都是str要么都是floatshow_all_types('a', 2.3)
上面内容基本上满足pydantic涵盖的泛型知识点.
pydantic 用法汇总
这里需要提出一个概念, 类型提示(type hints), pydantic 的schema校验是基于python类型提示的:
The schema that Pydantic validates against is generally defined by Python type hints.
其实python的类型提示指的就是指定变量、类属性或函数参数或返回值的预期类型的注解,和笔者前文说的类型注解type annotation只是叫法上不同.类型提示(type hints)可以看作语义上的叫法,类型注解(type annotation)可以看作语法上的叫法. 不必去纠结叫法问题.
pydantic本身主要功能就是定义schema,完成数据校验以及数据序列化,首先就是需要明白pydantic是怎么去定义数据的schema的,目前pydantic提供了4中方式去定义schema
schema的四种校验方式
数据模型类
继承自pydantic 的 BaseModel类定义一个数据模型,此模型则表示数据schema.
from typing import Annotated
from annotated_types import Le, Ge
from pydantic import BaseModel, ConfigDict, ValidationError# 继承BaseModel的类则是表示schema,这种类也被称做数据模型
class A(BaseModel):# 字段则是带类型注解的类属性,不带类型注解会报错name: strage: int# 配置这一部分命名一定是model_config,其他名字会报错# 这里配置为严格模式,字段值必须与类型完全匹配model_config = ConfigDict(strict=True)def test1():try:a = A(name='abc', age='11')except ValidationError as e:print(e)# 模型实例化过程即是解析验证过程,如果这个过程没有抛出异常,那么得到的实例就是满足模型定义的schemaa = A(name='abc', age=10)print(a.model_dump())# 其次创建一个模型实例传参必须是关键字参数,不支持位置参数传递# 这里运行会报错,静态分析工具也会提示错误#b = A('acb', 11)if __name__ == '__main__':test1()
输出
1 validation error for A
ageInput should be a valid integer [type=int_type, input_value='11', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_type
{'name': 'abc', 'age': 10}
model copy 模型实例拷贝
模型实例拷贝和python的一般对象拷贝类似,包含浅拷贝和深拷贝举个例子:
from pydantic import BaseModelclass Base(BaseModel):arg0: strclass Sub(BaseModel):arg1: intarg2: floatarg3: Basedef test():obj = Sub(arg1=1, arg2=2.3, arg3={'arg0': 'test'})# 拷贝同时还可以更新字段copy_obj = obj.model_copy(update={'arg2': 3.4})print(repr(copy_obj))# 浅拷贝print(id(copy_obj.arg3) == id(obj.arg3))# 深拷贝deep_copy_obj = obj.model_copy(deep=True)print(id(deep_copy_obj.arg3) == id(obj.arg3))
输出:
Sub(arg1=1, arg2=3.4, arg3=Base(arg0='test'))
True
False
目前此功能使用场景并不多.
model validate 基本的数据校验
pydantic提供了三个validate相关类方法分别是model_validate,model_validate_json以及model_validate_strings
首先是model_validate方法,适用场景是已经存在一个python字典想要校验得到模型实例
class B(BaseModel):name: str = Field(default='abc', min_length=1)age: Annotated[int, Ge(0), Le(100)]sign_time: datetime | None = Nonedef test2():# 这里obj 可以是python字典,也可以是对应模型的实例,和模型的__init__方法类似,只是传参不再是关键字参数# 由于这里不是严格模式,所以age字符串,sign_time字符串都能够被正常解析校验b = B.model_validate({'name': '123', 'age': '20', 'sign_time': '2025-10-09T22:00:00Z'})print(b)try:# 严格模式报错b1 = B.model_validate({'name': '123', 'age': '20', 'sign_time': '2025-10-09T22:00:00Z'}, strict=True)except Exception as e:print('='*40)print(e)tz_utc = ZoneInfo('UTC')# 严格模式不存在类型转换,类型必须完全匹配b2 = B.model_validate({'name': '123', 'age': 20, 'sign_time': datetime(2025,10,9,22,0,0,tzinfo=tz_utc)})print('='*40)print(b2)
输出:
name='123' age=20 sign_time=datetime.datetime(2025, 10, 9, 22, 0, tzinfo=TzInfo(UTC))
========================================
2 validation errors for B
ageInput should be a valid integer [type=int_type, input_value='20', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_type
sign_timeInput should be a valid datetime [type=datetime_type, input_value='2025-10-09T22:00:00Z', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/datetime_type
========================================
name='123' age=20 sign_time=datetime.datetime(2025, 10, 9, 22, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC'))
其次是model_validate_json:主要负责处理校验json数据或其编码字节数据场景.比如api调用返回json字符串,或者网络通信获取的字节码数据校验.
class B(BaseModel):name: str = Field(default='abc', min_length=1)age: Annotated[int, Ge(0), Le(100)]sign_time: datetime | None = Nonedef test3():# 对输入的json字符串或者其编码后的字节对象进行校验生成模型实例# 适合场景为:api调用获取的json数据校验;网络通信工具,库等(如socket)获取的json字节码数据校验b = B.model_validate_json('{"name": "abc", "age":10, "sign_time": "2025-01-01T10:00:00Z"}')print(b)# strict 为True场景,时间字符串不会报错,应为从json支持的数据类型来讲,它是合法的.try:b = B.model_validate_json('{"name": "abc", "age": "10", "sign_time": "2025-01-01T10:00:00Z"}', strict=True)print('='*40)print(b)except ValidationError as e:# 这里只报错age字段类型不匹配print('='*40)print(e)if __name__ == '__main__':test3()
输出:
name='abc' age=10 sign_time=datetime.datetime(2025, 1, 1, 10, 0, tzinfo=TzInfo(UTC))
========================================
1 validation error for B
ageInput should be a valid integer [type=int_type, input_value='10', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_type
第三个函数则是model_validate_strings这个函数接收的参数也是一个字典(可嵌套),但是这个字典的key和value都必须是字符串,且校验的时候按照json模式进行校验.校验过程中可以完成string类型值到模型定义的类型的转换.此函数其实是对model_validate功能的补充,举个例子:
class Innerc(BaseModel):arg: intclass C(BaseModel):name: str = Field(default='abc', min_length=1)age: Annotated[int, Ge(0), Le(100)]sign_time: datetime | None = Noneinner_c: Innerc | None = Nonevalid: bool = Falsedef test4():try:# 嵌套model初始化可以传递字典c = C(name='abc', age=10, inner_c={'arg': 1})print(c)except ValidationError as e:print('='*40)print(e)try:# model_validate 严格模式下字符串转换到int和bool会报错c = C.model_validate({'age': '10', 'valid': 'true'}, strict=True)print('='*40)print(c)except ValidationError as e:print('='*40)print(e)try:# model_validate_strings 严格模式下下字符串转换到int和bool不会报错,应为它本身要求输入的dict值类型都是stringc = C.model_validate_strings({'age': '10', 'valid': 'true'}, strict=True)print('='*40)print(c)except ValidationError as e:print('='*40)print(e)# json 字符串转换.c = C.model_validate_json('{"age":10, "valid": true}')print('='*40)print(c)if __name__ == '__main__':test4()
输出:
name='abc' age=10 sign_time=None inner_c=Innerc(arg=1) valid=False
========================================
2 validation errors for C
ageInput should be a valid integer [type=int_type, input_value='10', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_type
validInput should be a valid boolean [type=bool_type, input_value='true', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/bool_type
========================================
name='abc' age=10 sign_time=None inner_c=None valid=True
========================================
name='abc' age=10 sign_time=None inner_c=None valid=True
总之主流的校验函数还是model_validate和model_validate_json. model_validate_strings可作为model_validate的补充,当输入是一个字典且取值都是字符串的场景,考虑使用model_validate.
model dump 基本的模型序列化.
模型基本的序列化方法是两个model_dump和model_dump_json前者是序列化成一个python字典,后者是直接序列化成一个json字符串方便后续进行网络传输,其中需要注意的是model_dump的mode参数取值可为json或者python,这个参数不会影响model_dump输出字典类型,具体细节是model_dump(mode='python')(默认)输出的字典中的值是保留原始python类型,而model_dump(mode='json')是序列化的值是json兼容类型,其序列化结果使用json.dumps即可得到一个标准的json字符串,举例如下:
from datetime import datetime
from zoneinfo import ZoneInfoclass B(BaseModel, strict=True):name: Annotated[str, Field(min_length=1, max_length=10)]age: Annotated[int, Field(ge=0, le=100)]register_datetime: datetimedef test():input_params = {'name': 'test','age': 10,'register_datetime': datetime(year=2025, month=10, day=10, tzinfo=ZoneInfo('UTC'))}obj = B.model_validate(input_params)print(repr(obj))# model_dump_json# 返回json字符串json_res = obj.model_dump_json()print(type(json_res))# register_datetime 是iso 8601 标准的时间字符串print(json_res)python_dump = obj.model_dump()print(type(python_dump))# register_datetime 依然是python 的 datetime.datetimeprint(python_dump)json_dump = obj.model_dump(mode='json')print(type(json_dump))# register_datetime 是iso 8601 标准的时间字符串print(json_dump)if __name__ == '__main__':test()
输出结果:
B(name='test', age=10, register_datetime=datetime.datetime(2025, 10, 10, 0, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC')))
<class 'str'>
{"name":"test","age":10,"register_datetime":"2025-10-10T00:00:00Z"}
<class 'dict'>
{'name': 'test', 'age': 10, 'register_datetime': datetime.datetime(2025, 10, 10, 0, 0, tzinfo=zoneinfo.ZoneInfo(key='UTC'))}
<class 'dict'>
{'name': 'test', 'age': 10, 'register_datetime': '2025-10-10T00:00:00Z'}
总之序列化的结果还要继续在当前python执行环境使用选择model_dump且模式是默认的python;如果是需要将序列化结果通过网络发送到其他系统,果断选择model_dump_json.
generic model 泛型模型
pydantic模型基础上引入了泛型,基本用法如下:
from typing import Annotated, Generic, TypeVar
from pydantic import BaseModel, Field, ValidationErrorT = TypeVar('T')# 最基本的泛型模型
class Response(BaseModel, Generic[T]):data: Tdef test():# 泛型具体类型和相关字段传递数据类型不一致,但是这里可以转换(str-->int),所以不报错.res = Response[int](data='1')print(res)res = Response[str](data='abc')print(res)try:# 泛型具体类型和相关字段传递数据类型不一致,也不能进行转换,所以报错.res = Response[int](data='abc')except ValidationError as e:print(e)# 实例化的时候不显式指定类型参数的具体类型,也能够成功构建,但是还是推荐显式指定具体类型res = Response(data=1)print(res)if __name__ == '__main__':test()
输出:
data=1
data='abc'
1 validation error for Response[int]
dataInput should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_parsing
data=1
或者有具体类型的参数化泛型模型作为其它模型字段的类型注解:
class Product(BaseModel):id: intname: strclass Order(BaseModel):id: intproduct: Response[Product]def test():obj = Order(id=1, product={'data': {'id': 1, 'name': 'abc'}})print(repr(obj))# 或者product = Product(id=1, name='test')response_product = Response[Product](data=product)obj = Order(id=1, product=response_product)print(repr(obj))if __name__ == '__main__':test()
输出:
Order(id=1, product=Response[Product](data=Product(id=1, name='abc')))
Order(id=1, product=Response[Product](data=Product(id=1, name='test')))
不进行参数化的泛型模型校验问题,如下所示,这种情况下pydantic会依据类型变量定义时的边界或者约束集合中的类型去做校验.
T_normal = TypeVar('T_normal')
T_bound = TypeVar('T_bound', bound=int)
T_cons = TypeVar('T_cons', str, int)class NoParam(BaseModel, Generic[T_normal, T_bound, T_cons]):arg1: T_normalarg2: T_boundarg3: T_consdef test():# 规范实例化需要指定类型参数具体类型res = NoParam[int, int, str](arg1=123, arg2=123, arg3='123')print(repr(res))# 不显式指定类型校验会参考边界类型或者约束类型集合res = NoParam(arg1='any', arg2=123, arg3=3)print(repr(res))# 如果不能满足边界类型或者约束类型集合则报错try:res = NoParam(arg1='any', arg2=2.3, arg3=3.2)except ValidationError as e:print(e)
不进行参数化的泛型模型直接实例化还可能导致数据丢失,比如下面一个例子:
T_Item = TypeVar('T_Item', bound='ItemBase')class ItemBase(BaseModel):passclass IntItemBase(ItemBase):arg: intclass GenModel(BaseModel, Generic[T_Item]):item: T_Itemdef test():# 由于没有指定具体类型,pydantic这里按照类型变量定义时上边界类型去做校验# 所以obj对象中item就是空的obj = GenModel(item={'arg': 1})print(repr(obj))# 正确做法显式指定类型obj = GenModel[IntItemBase](item={'arg': 1})print(repr(obj))if __name__ == '__main__':test()
输出结果:
GenModel(item=ItemBase())
GenModel[IntItemBase](item=IntItemBase(arg=1))
总之泛型模型场景实例化以及校验的时候一定要明确指定具体类型,不然可能结果和预期的不同.无论是数据校验还是后续的序列化过程,遵循最规范的做法显式指定泛型类型参数的具体类型即可杜绝很多错误.
dynamic model 动态模型
前面的例子都是事先定义好模型,然后用模型来校验输入数据得到实例.实际上pydantic也提供了动态创建模型的函数,即是create_model函数.
动态模型创建的例子:
from pydantic import create_model, BaseModel, Field
from typing import Annotated# 第一个参数是模型名字,后续参数为字段定义
# 基本上的字段定义的语法如下
DynamicModel = create_model('DynamicModel',# 简单的name 字段类型为strname=str,# bool型字段,默认值为Falsevalidated=(bool, False),# age字段结合Field去定义age=(int, Field(ge=0, le=100)),# 使用Annotated去定义字段语法nickname=Annotated[str, Field(min_length=1, max_length=20)]
)# 等价的静态模型定义
class StaticModel(BaseModel):name: strvalidated: bool = Falseage: int = Field(ge=0, le=100)nickname: Annotated[str, Field(min_length=1, max_length=20)]def test():raw_data = {'name': 'a','validated': True,'age': 10,'nickname': 'test'}obj1 = DynamicModel.model_validate(raw_data)print(obj1.model_dump())obj2 = StaticModel.model_validate(raw_data)print(obj2.model_dump())if __name__ == '__main__':test()
输出:
{'name': 'a', 'validated': True, 'age': 10, 'nickname': 'test'}
{'name': 'a', 'validated': True, 'age': 10, 'nickname': 'test'}
动态创建模型也可以继承静态模型,例子如下:
class BaseClass(BaseModel):arg1: intSubClass = create_model('SubClass',name=str,__base__=BaseClass
)def test():obj = SubClass.model_validate({'name': 'abc', 'arg1': 1})print(repr(obj))if __name__ == '__main__':test()
输出:
SubClass(arg1=1, name='abc')
实际使用过程中,动态创建模型场景还是相对较少.
模型字段与类型详解
首先罗列一下所有定义字段的方式,然后总结最规范化的实现.首先看下面一个model的定义代码.
from pydantic import Field, BaseModel, ValidationError, PositiveFloat
from typing import Annotated
from annotated_types import MinLen, MaxLenclass MyModel(BaseModel):# 基本类型,无任何约束,直接使用类型注解id: int# Filed赋值给属性,这里语法上是赋值,实际上是pydantic能够识别的给字段加上约束的行为# 这种语法适合Field就能满足给字段添加约束的场景,如果Field不能满足则不能使用这种语法# 其次这种语法从python语法层面即是赋值操作,某些代码类型检查工具可能会认为这里有错# 然后这种语法也会产生误解,认为字段已经赋值,有默认值了,产生实例化时就不用赋值的错误理解,如果要设置默认值使用Field(default=xxx)# 综合考虑,Field赋值语法适合简单类型,且约束条件Field能够cover的场景name: str = Field(min_length=1)# 最推荐的字段定义方式 使用 Annotated + Field提供约束/元数据,如果Field还不足够,它还能支持添加更多的元数据scores: list[Annotated[float, Field(ge=1.0, le=10.0)]]# Annotated 还可以添加annotated_types下面的元数据/约束类等,是最推荐的定义有约束的字段的方式alias: Annotated[str, MinLen(1), MaxLen(10)]# 同时还可以使用pydantic.types 模块下预定意的类型,比如PositiveFloat,其本质上也是Annotated + annotated_types中的约束height: PositiveFloatdef test():try:obj = MyModel(id=1, name='abc', scores=[-1, 1, 2], alias='test', height=173)except ValidationError as e:print(e)if __name__ == '__main__':test()
输出结果:
1 validation error for MyModel
scores.0Input should be greater than or equal to 1 [type=greater_than_equal, input_value=-1, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/greater_than_equal
所以综合考虑,定义字段最规范化的实现是:
- 如果是不带任何约束/元数据类型直接使用
attr: type这种方式定义字段,记得类型要写正确. - 如果是需要有约束,优先看pydantic是否已经预定以了所需类型,直接
attr: pydantic_type即可.如果没有预定义类型且约束条件Field能完全cover,则选择Field赋值语法 - 上面两种情况都不能满足,考虑使用Annotated + Field + 其他元数据的方式定义字段,这种定义的语法是灵活度最高的,同时基本能cover所有字段定义的场景.
查看模型字段使用model_fields类属性,返回的是一个字典,字典key为字段名(注意非别名),字段值是FieldInfo类型,可以查看具体字段的信息(比如metadata),举个例子:
class TestFields(BaseModel):name: str = Field(default='abc', frozen=True)def test():fields = TestFields.model_fieldsprint(type(fields))for k, v in fields.items():print(k, repr(v))if __name__ == '__main__':test()
输出:
<class 'dict'>
name FieldInfo(annotation=str, required=False, default='abc', frozen=True)
默认值问题,目前三种字段默认值设置方式.直接赋值,Field(default=xxx)设置default value 或者使用Field(default_factory=callable) 传递一个可调用对象实现值的初始化.
class DefaultModel(BaseModel):# 直接赋值id: int = 1# Field中default参数name: str = Field(default=' abc ')# 细节Field(default=xxx) 和 赋值同时存在,最终默认值是赋值操作给的值,Field中default无效,不建议使用这种语法# 还是使用Field中的default参数设置初始值age: Annotated[int, Field(default=1)] = 10# 使用default_factory 传递callable 无参数seq: str = Field(default_factory=lambda: uuid.uuid4().hex)# 使用default_factory 但是带单个参数,注意这里data中是验证过的字段,且需要考虑字段定义顺序# 选择name字段是name在clean_name之前就已经验证了,如果没有验证这里会出错clean_name: str = Field(default_factory=lambda data: data['name'].strip())def test():model = DefaultModel()print(repr(model))if __name__ == '__main__':test()
输出:
DefaultModel(id=1, name=' abc ', age=10, seq='5794116e03c94ee7bd69c774b8ae1e63', clean_name='abc')
展开讲字段别名问题.别名是一个及其鸡肋的功能,完全是增加开发负担的东西,举个例子:
class AliasDemo(BaseModel):# 校验以及序列化都是别名id: int = Field(alias='id_als')# 校验别名name: str = Field(validation_alias='name_val')# 序列化别名age: int = Field(serialization_alias='age_ser')def test():# 此处创建对象id参数必须用别名, name参数既可以字段名,又可以别名, age参数只能字段名字obj = AliasDemo(id_als=1, name_val='2', age=3)# repr展示的全是字段名print(repr(obj))# 输出字段名print(list(AliasDemo.model_fields.keys()))# id 和 age输出别名print(obj.model_dump_json(by_alias=True))if __name__ == '__main__':test()
输出:
AliasDemo(id=1, name='2', age=3)
['id', 'name', 'age']
{"id_als":1,"name":"2","age_ser":3}
会看到一会儿是真实字段名,一会儿又是别名,一会儿真实字段名别名都可以用. 完全是积累的东西.这里只要满足规范要求,字段名命名意义明确,根本不需要什么别名,完全多此一举,增加开发负担.
计算字段之computed_field装饰器
from functools import cached_property
import json
class ComputeModel(BaseModel):arg1: intarg2: intarg3: int@computed_field@propertydef func1(self) -> int:return self.arg1 * self.arg2# cached_property只计算一次,且属性是只读的@computed_field@cached_propertydef func2(self) -> int:return self.arg2 * self.arg3def test():# 计算字段不会出现在model_fields里面,本质还是实例属性print(list(ComputeModel.model_fields.keys()))obj = ComputeModel(arg1=1, arg2=2, arg3=3)# 当实例属性访问print(obj.func1, obj.func2)# 初始化之后 @property 属性不可赋值, @cached_property可以赋值,但一般不这么做#obj.func1 = 10#obj.func2 = 20# 序列化结果里面计算字段会出现print(obj.model_dump())# serialization 模式下的json schema 有 计算字段print(json.dumps(ComputeModel.model_json_schema(mode='serialization'), indent=2))if __name__ == '__main__':test()
输出结果:
['arg1', 'arg2', 'arg3']
2 6
{'arg1': 1, 'arg2': 2, 'arg3': 3, 'func1': 2, 'func2': 6}
{"properties": {"arg1": {"title": "Arg1","type": "integer"},"arg2": {"title": "Arg2","type": "integer"},"arg3": {"title": "Arg3","type": "integer"},"func1": {"readOnly": true,"title": "Func1","type": "integer"},"func2": {"readOnly": true,"title": "Func2","type": "integer"}},"required": ["arg1","arg2","arg3","func1","func2"],"title": "ComputeModel","type": "object"
}
模型的 json schema
回到模型表达的schema问题.模型本身存在model_json_schema方法返回一个可json化的字典,代表模型包含的schema信息.生成的字典可以直接通过json.dumps即是json.dumps(m.model_json_schema())生成json schema字符串.如下一个最基本的json schema用法例子:
from pydantic import BaseModel, ValidationError, Field, WithJsonSchema
from typing import Annotated
import jsonclass MySchema(BaseModel, strict=True):id: int# Field中和schema相关的参数title, description, examples 以及 json_schema_extraname: str = Field(title='CName', description='a test name description', examples=['abc'])# json_schema_extra 是追加额外的schema信息,可以覆盖title, description, examples设置的值age: Annotated[int, Field(title='The Age', json_schema_extra={'title': 'new age title','description': 'age description'})]# WithJsonSchema 元数据类也可以定制化字段schema,但是他是完全覆盖默认的字段schema中除了title字段之外的所有信息# 使用它必须给出完整的定制化schema内容.# 比如这里必须要给出type: integer, 不然生成schema这里会缺失typeheight: Annotated[int, WithJsonSchema({'type': 'integer', 'description': 'height value'})]def test():# 两种mode 默认是 'validation' 还可以取值'serialization'# 返回一个可json化的字典(jsonable dict)schema = MySchema.model_json_schema(mode='validation')# json.dumps 即可以得到json schema stringprint(json.dumps(schema, indent=2))if __name__ == '__main__':test()
输出:
{"properties": {"id": {"title": "Id","type": "integer"},"name": {"description": "a test name description","examples": ["abc"],"title": "CName","type": "string"},"age": {"description": "age description","title": "new age title","type": "integer"},"height": {"description": "height value","title": "Height","type": "integer"}},"required": ["id","name","age","height"],"title": "MySchema","type": "object"
}
下面是模型级别的json schema定制化的细节
首先模型级别的schema定制化可以在两个地方配置实现. 第一个是模型的配置参数model_config创建时给ConfigDict相应参数赋值;第二个是模型定义的时的类关键字参数,且同时使用两种配置方式且设置同样的参数的时候,第二种方式优先级更高.但是笔者认为第一种配置方式更加规范.
class ModelLevelSchema(BaseModel, title='abc'):id: int# 更推荐这种配置方式定制化model schemamodel_config = ConfigDict(title='bcd')def test():schema = ModelLevelSchema.model_json_schema()print(json.dumps(schema, indent=2))if __name__ == '__main__':test()
输出:
{"properties": {"id": {"title": "Id","type": "integer"}},"required": ["id"],"title": "abc","type": "object"
}
总之模型级别的json schema定制话最佳实现就是在模型的model_config特殊配置属性里面配置schema相关的参数.模型级别的json schema 定制化并不常见,虽然官方文档https://docs.pydantic.dev/latest/concepts/json_schema/#model-level-customization 这一部分写了不少,但是都比较鸡肋,实际开发这些功能使用并不多,这里就不再赘述.
customized validator 定制化校验器
pydantic支持定制化校验器进行更复杂的校验,分为字段级别的校验器和模型级别校验器
字段级别校验器
字段级别校验器有四种类型(BeforeValidator, AfterValidator,PlainValidator以及WrapValidator),每种类型有两种语法(annotated pattern以及field_validator装饰器),
后置(After)校验器
后置校验逻辑是在pydantic完成内置校验之后执行.所以这个校验逻辑的输入参数类型以及输出结果类型都应当和对应字段类型一致.
annotated pattern语法实现
def is_even(value: int) -> int:if value % 2 == 1:raise ValueError(f'such input: {value} is not event')return valueclass AfterVA(BaseModel):name: Annotated[str, ...]number: Annotated[int, AfterValidator(is_even)]model_config = ConfigDict(strict=True)def test():try:obj = AfterVA(name=1, number=1)except ValidationError as e:print(e)if __name__ == '__main__':test()
输出
2 validation errors for AfterVA
nameInput should be a valid string [type=string_type, input_value=1, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/string_type
numberValue error, such input: 1 is not event [type=value_error, input_value=1, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/value_error
field_validator装饰器实现版本,装饰器必须装饰在类方法上且对参数类型要求和annotated pattern模式一致.这里装饰器的第一个参数是目标字段,取值是字段名字符串,可以是一个也可以是多个,如果赋值*则是表示所有字段都要用当前校验函数进行校验
class AfterVB(BaseModel):name: Annotated[str, ...]number: intmodel_config = ConfigDict(strict=True)# 类方法上修饰# 第一个参数可以是一个字段名,多个字段名,或者* 全部字段是用@field_validator('number', mode='after')@classmethoddef is_even(cls, value: int) -> int:if value % 2 == 1:raise ValueError(f'such input: {value} is not event')return valuedef test():try:obj = AfterVB(name=1, number=1)except ValidationError as e:#两个字段错误都正确打印print(e)if __name__ == '__main__':test()
输出:
nameInput should be a valid string [type=string_type, input_value=1, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/string_type
numberValue error, such input: 1 is not event [type=value_error, input_value=1, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/value_error
小结:AfterValidator只能够在前置校验全部通过后才会运行校验逻辑,如果前面的校验都无法通过就不可能运行此函数.
前置(Before)校验器
前置校验输入数据类型是Any输出类型也是Any,输出数据就进入pydantic 内置校验流程.它本身是进行预处理的验证器,它的核心作用是转换而非验证.可以在原始输入类型之上预先进行一次目标字段类型转换的操作,如果转换不了就直接抛出异常,这样后续校验逻辑不用进行,是一种比较高效的处理逻辑.比如下面一个annotated pattern实现版本:
# BeforeValidator的校验函数基本入参和输出都是Any
def is_number(value: Any) -> Any:# 这里定义字段需要的是int 类型,这里可以尝试先转换,如果不行直接抛出异常,就不需要执行后续校验过程if not isinstance(value, int):try:return int(value)except Exception:raise# 类型满足就直接return,进入内置校验器验证阶段return valueclass BeforeVA(BaseModel):number: Annotated[int, BeforeValidator(is_number)]def test():try:obj = BeforeVA(number='acb')except ValueError as e:print(e)if __name__ == '__main__':test()
输出:
1 validation error for BeforeVA
numberValue error, invalid literal for int() with base 10: 'acb' [type=value_error, input_value='acb', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/value_error
field_validator版本:
class BeforeVB(BaseModel):number: int@field_validator('number', mode='before')@classmethoddef is_number(cls, value: Any) -> Any:if not isinstance(value, int):try:return int(value)except Exception:raisereturn valuedef test():try:obj = BeforeVB(number='acb')except ValueError as e:print(e)if __name__ == '__main__':test()
输出:
1 validation error for BeforeVB
numberValue error, invalid literal for int() with base 10: 'acb' [type=value_error, input_value='acb', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/value_error
普通(Plain)验器
普通验器其输入输出和BeforeValidator一致,但是比它更积累,只要这个验证器运行且返回值,那么这个值就是模型字段最终取值,模型内置以及后续校验逻辑不在运行.总之意义不是很大.下面是annotated pattern语法版本的例子.
def double_value(value: Any) -> Any:if isinstance(value, int):return value * 2else:return valueclass PlainVA(BaseModel):id: Annotated[int, PlainValidator(double_value)]model_config = ConfigDict(strict=True)def test():# 内置校验器不再运行,这里类型不匹配也能初始化不报错obj = PlainVA(id='ssss')print(repr(obj))if __name__ == '__main__':test()
输出:
PlainVA(id='ssss')
field_validator版本:
class PlainVB(BaseModel):id: int@field_validator('id', mode='plain')@classmethoddef double_value(cls, value: Any) -> Any:if isinstance(value, int):return value * 2else:return valuedef test():obj = PlainVB(id='ssss')print(repr(obj))if __name__ == '__main__':test()
输出:
PlainVB(id='ssss')
包装(Wrap)校验器
包装校验器功能最灵活的校验器,可以在其它校验逻辑前处理原始输入,或者在其它校验逻辑之后再处理字段数据,甚至可以捕获校验异常进行后续加工处理,功能是四个教研器中最强大的.例子如下:
def truncate(value: Any, handler: ValidatorFunctionWrapHandler) -> str:try:# 其它校验器之前的逻辑,没有强制要求调用handler,在调用之前返回代表不再进行其它校验res = handler(value)# 其它校验 之后的逻辑 可以直接返回handler的值return resexcept ValidationError as err:# 捕获整个校验过程中的错误# err.errors() 返回整个校验过程中的错误if err.errors()[0]['type'] == 'string_too_long':# 这类错误则直接处理,不会抛出异常return handler(value[:5])else:raiseclass WrapVA(BaseModel):name: Annotated[str, WrapValidator(truncate)]model_config = ConfigDict(strict=True)def test():# 不抛出异常,wrap validator 会完成字符串截取操作obj = WrapVA(name='abcdefg')try:# 抛出异常,因为wrap validator 校验后处理会直接抛出这种类型不匹配的错误.obj = WrapVA(name=123)except ValidationError as e:print(e)if __name__ == '__main__':test()
输出结果:
1 validation error for WrapVA
nameInput should be a valid string [type=string_type, input_value=123, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/string_type
field_validator版本:
class WrapVB(BaseModel):name: strmodel_config = ConfigDict(strict=True)@field_validator('name', mode='wrap')@classmethoddef truncate(cls, value: Any, handler: ValidatorFunctionWrapHandler) -> str:try:res = handler(value)return resexcept ValidationError as err:if err.errors()[0]['type'] == 'string_too_long':return handler(value[:5])else:raisedef test():obj = WrapVB(name='abcdefg')try:obj = WrapVB(name=123)except ValidationError as e:print(e)if __name__ == '__main__':test()
输出:
1 validation error for WrapVB
nameInput should be a valid string [type=string_type, input_value=123, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/string_type
自定义校验器异常的细节.
自定义校验器需要抛出异常的场景,官方推荐三种ValueError,AssertionError以及PydanticCustomError. 第二种不推荐,它是assert xxx不为True抛出的异常,但是如果运行python脚本时使用了-O参数它将不生效,所以开发过程中需要抛出异常时选择其它两种异常类型.其中ValueError是最基本的异常类型,而PydanticCustomError继承自它,可以提供更丰富的异常信息,它支持自定义异常类型(字符串),支持异常信息模板以及异常信息上下文(可选).但是无论是那种异常最终都会被封装到ValidationError中并且可以通过访问此对象中的errors列表获取详细的校验异常数据.举个例子:
class ErrorTypeV(BaseModel):number1: intnumber2: intname: str@field_validator('name', mode='after')@classmethoddef is_str_ok(cls, value: str) -> str:if len(value) >= 5:raise ValueError('str error', 'len > 5')return value@field_validator('number1', 'number2', mode='after')@classmethoddef is_number_ok(cls, value: int) -> int:if value < 5:# 无contextraise PydanticCustomError('NumberNoContext','number error no context')elif value > 5:# 有contextraise PydanticCustomError('NumberContext','{number} has context',{'number': value})return valuedef test():try:ErrorTypeV(number1=2, number2=6, name='abcdefrg')except ValidationError as e:# 校验错误的数量print(f'error count: {e.error_count()}')for err in e.errors():# 每一个error 类型都是dictprint(err)
if __name__ == '__main__':test()
输出:
error count: 3
{'type': 'NumberNoContext', 'loc': ('number1',), 'msg': 'number error no context', 'input': 2}
{'type': 'NumberContext', 'loc': ('number2',), 'msg': '6 has context', 'input': 6, 'ctx': {'number': 6}}
{'type': 'value_error', 'loc': ('name',), 'msg': "Value error, ('str error', 'len > 5')", 'input': 'abcdefrg', 'ctx': {'error': ValueError('str error', 'len > 5')}, 'url': 'https://errors.pydantic.dev/2.11/v/value_error'}
可见只要是抛出的ValueError,其type都是value_error,如果是PydanticCustomError 则type是可以自定义的,这样如果后续需要对异常类型进行统计那么选择抛出PydanticCustomError是更好的选择,如果只是为了打印异常信息那么也可以抛出简单的ValueError类型.
ValidtionInfo可选参数
每一个自定义校验器无论是annotated pattern语法还是field_validator语法,都支持在校验器函数最后面添加一个ValidationInfo可选参数.且这个参数还可以用于模型校验场景.这个参数对复杂的字段校验逻辑还是有比较大的帮助.使用这个参数主要关心其5个属性分别是:
- field_name(str): 当前校验的字段名字,
只有在字段级别校验场景有效,模型级别校验场景为None - config(dict): 模型的配置信息,不为空,至少存在一个
title属性. - mode(str): 模型校验模式,只有两个取值(
json/python),比较关键的参数.受到校验方式的影响.如果是调用初始化函数创建model(Model(xxx=xxx)以及使用Model.model_validate函数校验则mode取值为python,如果使用的是Model.model_validate_json则mode取值为json. - data(dict):
只在字段级别校验场景生效,表示当前已经完成校验的字段和其值,其默认值为空的字典({}).模型级别校验场景这个属性为None这个属性适合需要进行字段联合校验的场景.但是需要注意字段校验顺序,字段校验顺序和定义顺序一致,一定要保证当前字段中选取其它字段的时候,其它字段已经弯成校验,不然data中没有选择字段的值. - context(dict): 校验的上下文,
默认为None,只有在执行校验的时候显式传递context(如Model.model_validate(context-xxx))才能在校验过程中获取上下文信息.
举个例子:
class ValidationInfoV(BaseModel):field1: intfield2: str@field_validator('field1', mode='wrap')@classmethoddef field1_validate(cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo) -> int:print(info)print(info.field_name, info.mode, info.config)if info.context:print(info.context)print(info.data)return handler(value)@field_validator('field2', mode='wrap')@classmethoddef field2_validate(cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo) -> str:# info 细节print(info)print(info.field_name, info.mode, info.config)if info.context:print(info.context)print(info.data)return handler(value)def test():# mode = python, context = Noneobj = ValidationInfoV(field1=1, field2='abc')# mode - json, context = {'a': 1}obj = ValidationInfoV.model_validate_json('{"field1": 1, "field2": "abc"}', context={'a': 1})
模型级别校验器
模型级别的校验器只有三种后置校验器,前置校验器以及包装校验器. 但是模型级别的校验器只有使用装饰器@model_validator语法,没有annotated pattern的语法.同时模型级别的校验器也可以传人ValidationInfo实例,只是相比于字段级别校验器,其data和field_name属性都为None.
后置(After)校验器
模型级别后置校验器和子度级别校验器有区别,因为后置校验器是在整个模型上校验,而且它是在模型所有字段都通过内置校验器之后才运行,所以这种场景下的装饰器装饰的是普通方法(实例方法)返回实例本身,举例如下:
class AfterV(BaseModel):name: strage: int# 只有mode参数 且装饰器装饰的函数必须返回实例本身@model_validator(mode='after')def model_validate_demo(self, info: ValidationInfo) -> Self:# model 校验中 ValidationInfo 的 data 和 field_name 都是 None,几乎使用不到这个两个属性print(info)if self.age > 100:raise ValueError('age', 100)return selfdef test():#obj = AfterV(name='abc', age=10)try:obj = AfterV(name=1, age=10)except ValidationError as e:print(e)try:obj = AfterV(name='abc', age=101)except ValidationError as e:print(e)
模型后置校验器相比于字段后置校验器更适合做联合字段校验,因为函数参数self代表整个内置校验完毕的实例,在这个实例上就可以选取任意字段,还不用考虑字段校验顺序.
前置(Before)校验器
前置校验器是在整个内置校验逻辑之前进行,此时的实例尚不存在,所以这种场景下的校验器必须作用于类方法,而且它类似于字段的前置校验器,只是作转换而不是验证.它的输入输出都是Any,举个例子:
class BeforeV(BaseModel):name: strage: int@model_validator(mode='before')@classmethoddef model_validate_demo(cls, data: Any, info: ValidationInfo) -> Any:# data 类型问题 多半是dict类型 不管mode是python 还是 jsonprint(type(data))if isinstance(data, dict):if 'test' in data:raise ValueError('error')return datadef test():# mode pythonobj = BeforeV(name='abc', age=10)# mode jsonobj = BeforeV.model_validate_json('{"name": "abc", "age": 10}')
包装(Wrap)校验器
同样,模型级别包装校验器和字段级别包装校验器类似也是功能最灵活最强大的校验器,支持在其它校验器前后添加逻辑,函数输入数据类型是Any输出是类型本身也就是Self,值得注意的是和字段的包装校验器类似,需要一个handler,只不过类包装校验器的handler类型为ModelWrapValidatorHandler[Self],举个例子:
class WrapV(BaseModel):name: strage: int@model_validator(mode='wrap')@classmethoddef model_validate_demo(cls, data: Any, handler: ModelWrapValidatorHandler[Self]) -> Self:try:print('wrap validator runs')res = handler(data)return resexcept ValidationError as e:raise edef test():obj = WrapV(name='acb', age=10)
customized serializer 定制序列化
和定制化validators类似,pydantic 也支持定制化的serializer.同样也支持字段级别的serializer 以及 模型级别的serializer.但是基本上只有两类别:PlainSerializer以及WrapSerializer.字段级别的序列化器,语法支持annotated pattern以及@field_serializer装饰器.
字段级别序列化器
普通(Plain)序列化器
默认情况下内置序列化产生的数据类型是dict[str, object],而普通序列化器可以根据需求指定返回的序列化结果的内容.比如下面的例子可以实现:
def set_number(value: Any) -> Any:print(type(value))if isinstance(value, int):return value * 2else:return valueclass PlainSA(BaseModel):number: Annotated[int, PlainSerializer(set_number)]def test():obj = PlainSA(number=2)print(obj.model_dump_json(indent=2))# 默认赋值不触发校验过程obj.number = 'abc'print(obj.model_dump_json(indent=2))if __name__ == '__main__':test()
输出:
<class 'int'>
{"number": 4
}
<class 'str'>
{"number": "abc"
}
对应装饰器版本field_serializator:
class PlainSB(BaseModel):number: int# 第一个参数是字段名字,可以多个字段名,或者*表示该序列化器用于所有字段@field_serializer('number', mode='plain')def set_number(self, value: Any) -> Any:print(type(value))if isinstance(value, int):return value * 2else:return valuedef test():obj = PlainSB(number=2)print(obj.model_dump_json(indent=2))obj.number = 'abc'print(obj.model_dump_json(indent=2))if __name__ == '__main__':test()
输出:
<class 'int'>
{"number": 4
}
<class 'str'>
{"number": "abc"
}
包装(Wrap)序列化器
和字段校验器中的包装校验器类似,可在其他序列化逻辑前与后可以执行序列化逻辑,需要SerializerFunctionWrapHandler类型handler,举个例子:
def set_name(value: Any, handler: SerializerFunctionWrapHandler) -> Any:print('before inner serializer')res = handler(value)print('after inner serializer')return resclass WrapSA(BaseModel):name: Annotated[str, WrapSerializer(set_name)]def test():obj = WrapSA(name='1234')print(obj.model_dump_json(indent=2))if __name__ == '__main__':test()
输出:
before inner serializer
after inner serializer
{"name": "1234"
}
对应field_serializer装饰器版本:
class WrapSB(BaseModel):name: str@field_serializer('name', mode='wrap')def set_name(self, value: Any, handler: SerializerFunctionWrapHandler) -> Any:print('before inner serializer')res = handler(value)print('after inner serializer')return resdef test():obj = WrapSB(name='1234')print(obj.model_dump_json(indent=2))if __name__ == '__main__':test()
输出:
before inner serializer
after inner serializer
{"name": "1234"
}
FieldSerializationInfo 细节部分
序列化部分SerializetionInfo有两个类别:SerializationInfo和FieldSerializationInfo.后者是前者的子类,后者增加了field_name属性,专门处理字段级别序列化场景,前者主要是处理模型级别序列化场景.其主要的三个属性field_name,mode以及context和ValidationInfo一样,其它属性则是序列化函数调用时传递的属性,举个例子:
class SerialInfoDemo(BaseModel):sign_datetime: datetime@field_serializer('sign_datetime', mode='wrap')def set_name(self, value: Any, handler: SerializerFunctionWrapHandler, info: FieldSerializationInfo) -> Any:print(info)# mode: python 时 handler 返回类型是datetime# mode: json 时handler 返回类型是 datetime的 iso 8601 格式时间字符串# mode 会影响序列化之后的数据类型,所以实际自定义序列化器,需要参考mode参数. res = handler(value)print(type(res))return resdef test():obj = SerialInfoDemo(sign_datetime=datetime.now(tz=ZoneInfo('UTC')))# mode: pythonobj.model_dump()# mode: jsonobj.model_dump(mode='json')if __name__ == '__main__':test()
输出:
SerializationInfo(include=None, exclude=None, context=None, mode='python', by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, serialize_as_any=False)
<class 'datetime.datetime'>
SerializationInfo(include=None, exclude=None, context=None, mode='json', by_alias=False, exclude_unset=False, exclude_defaults=False, exclude_none=False, round_trip=False, serialize_as_any=False)
<class 'str'>
模型级别序列化器
同样也只有两种序列化器, Plain以及Wrap. 语法上只有model_serializer装饰器.模型序列化器中使用的是SerializationInfo类型的序列化信息类,相比于字段序列化中使用的FieldSerializationInfo类型少了field_name属性.
普通(Plain)序列化器
class PlainS(BaseModel):name: strsign_datetime: datetime# 装饰器只有一个参数mode, plain/wrap# 返回值类型问题,内置的序列化返回类型是dict[str, object],但是自定义的序列化器可以根据需求返回类型@model_serializer(mode='plain')def model_serial_func(self, info: SerializationInfo) -> str:print(info.mode)return f'{self.name} - {self.sign_datetime}'def test():obj = PlainS(name='abc', sign_datetime=datetime.now(tz=ZoneInfo('UTC')))print(obj.model_dump_json(indent=2))if __name__ == '__main__':test()
输出:
json
"abc - 2025-10-25 13:52:57.601644+00:00"
包装(Wrap)序列化器
包装序列化器使用SerializerFunctionWrapHandler类型handler,这个和字段的包装序列化handler类型一致.
class WrapS(BaseModel):name: strsign_datetime: datetime# 返回类型,标准的序列化过程是dict[str, object]@model_serializer(mode='wrap')def model_serial_func(self, handler: SerializerFunctionWrapHandler, info: SerializationInfo) -> dict[str, object]:print(info.mode)res = handler(self)res['fields'] = list(res)return resdef test1():obj = WrapS(name='abc', sign_datetime=datetime.now(tz=ZoneInfo('UTC')))print(obj.model_dump())if __name__ == '__main__':test()
输出:
python
{'name': 'abc', 'sign_datetime': datetime.datetime(2025, 10, 25, 13, 56, 8, 370494, tzinfo=zoneinfo.ZoneInfo(key='UTC')), 'fields': ['name', 'sign_datetime']}
pydantic.dataclasses.dataclass 修饰类定义schema
pydantic,dataclasses 提供了dataclass装饰器用于定义数据的schema, 这是使用pydantic的model之外的定义schema之外的一种方式,但是官方文档这一部中又把一些非规范化的实现加入进来导致阅读起来稍显混乱,这里笔者从规范化的实现角度讲解清楚这种首先方式,如下使用pydantic.dataclasses.dataclass装饰器的标准实现.
from pydantic.dataclasses import dataclass
from pydantic import Field, ValidationError, TypeAdapter
from typing import Annotated
from datetime import datetime@dataclass
class User:id: intage: Annotated[int, Field(ge=0, le=100)]sign_datetime: datetimedef test():# 调用初始化函数不要求参数全部为关键字参数,使用位置参数和关键字参数皆可# 非严格模式下也能完成类型转换.user = User('1', 100, sign_datetime='2025-01-01T10:00:00Z')print(repr(user))# 类型不匹配依然抛出异常try:user = User('abc', 100, 'abc')except ValidationError as e:print(e)# 类本身不像model有各种validate, dump的方法,需要用TypeAdapter进行包装user_adapter = TypeAdapter(User)obj = user_adapter.validate_python({'id': '1', 'age': 10, 'sign_datetime': '2025-01-01T10:00:00Z'})print(type(obj))# 这里dump_json直接返回bytes类型 和 model_dump_json 返回字符串稍微不同print(user_adapter.dump_json(obj))if __name__ == '__main__':test()
输出结果:
User(id=1, age=100, sign_datetime=datetime.datetime(2025, 1, 1, 10, 0, tzinfo=TzInfo(UTC)))
2 validation errors for User
0Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_parsing
2Input should be a valid datetime or date, input is too short [type=datetime_from_date_parsing, input_value='abc', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/datetime_from_date_parsing
<class '__main__.User'>
b'{"id":1,"age":10,"sign_datetime":"2025-01-01T10:00:00Z"}'
可见通过调用初始化方法创建实例和模型类似,也要经历类型转换以及校验过程.同时装饰类需要经过TypeAdapter包装才具有各种校验以及序列化的方法.配置部分,使用pydantic.dataclasses.dataclass装饰的类型有两种方式添加配置,一个是装饰器的config参数,另一种是类中__pydantic_config__属性,传值都是ConfigDict实例.举个例子
# validate_assignment=True 对象创建后赋值也会进行校验
@dataclass(config=ConfigDict(validate_assignment=True))
class ConfOne:id: int@dataclass
class ConfTwo:id: int__pydantic_config__ = ConfigDict(validate_assignment=True)def test():one = ConfOne(1)two = ConfTwo(2)try:one.id = 'abc'except ValidationError as e:print(e)try:two.id = 'abc'except ValidationError as e:print(e)if __name__ == '__main__':test()
输出:
1 validation error for ConfOne
idInput should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_parsing
1 validation error for ConfTwo
idInput should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='abc', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/int_parsing
extra部分,pydantic.dataclasses.dataclass装饰类型行为会有所不同.如下例子:
@dataclass(config=ConfigDict(extra='allow'))
class ExtraDemo:id: intname: strdef test():demo = ExtraDemo(id=1, name='abc', arg=3)# 没有arg3print(repr(demo))# 没有 __pydantic_extra__ 属性print(hasattr(demo, '__pydantic_extra__'))demo_type = TypeAdapter(ExtraDemo)res = demo_type.validate_python({'id': 1, 'name': 'abc', 'arg': 3})# 没有arg3print(repr(res))# 没有 __pydantic_extra__ 属性print(hasattr(res, '__pydantic_extra__'))if __name__ == '__main__':test()
输出:
ExtraDemo(id=1, name='abc')
False
ExtraDemo(id=1, name='abc')
False
可见extra配置在pydantic.dataclasses.dataclass中即便配置为allow也不生效.
同样dataclass装饰器装饰的类也可以自定义validator 和 serializer,例子如下:
@dataclass
class ValSer:name: strage: int@field_validator('name', mode='after')@classmethoddef name_val(cls, value: str) -> str:return value[:3]@field_serializer('age', mode='plain')def set_age(self, value: Any) -> Any:if isinstance(value, int):return value * 2return valuedef test():obj = ValSer(name='abcde', age=10)type_val = TypeAdapter(ValSer)print(type_val.dump_json(obj, indent=2).decode('utf-8'))if __name__ == '__main__':test()
输出:
{"name": "abc","age": 20
}
TypeAdapter定义schema
TypeAdapter本身就是解决非Model类型的校验,序列化以及json schema 功能不全的场景,利用TypeAdapter可以赋予这些类型相应的功能.前面提到的给pydantic.dataclasses.dataclass装饰器类赋予校验,序列化以及生成json schema能力,比如另外一个例子,对list[Model]类增加校验相关功能.
from typing_extensions import TypedDictclass User(TypedDict):id: intname: Annotated[str, Field(max_length=3)]age: Annotated[int, Ge(0), Le(100)]def test():# 不进行校验,非Model以及非pydantic.dataclasses.dataclass装饰类型obj = User(id=1, name='123456', age=101)print(repr(obj))user_type = TypeAdapter(User)try:obj = user_type.validate_python({'id': 1, 'name': 'abcdef', 'age': 101})except ValidationError as e:print(e)obj = user_type.validate_python({'id': 1, 'name': 'abc', 'age': 67})# raw dump 结果是bytesprint(user_type.dump_json(obj).decode('utf-8'))if __name__ == '__main__':test()
输出:
{'id': 1, 'name': '123456', 'age': 101}
2 validation errors for User
nameString should have at most 3 characters [type=string_too_long, input_value='abcdef', input_type=str]For further information visit https://errors.pydantic.dev/2.11/v/string_too_long
ageInput should be less than or equal to 100 [type=less_than_equal, input_value=101, input_type=int]For further information visit https://errors.pydantic.dev/2.11/v/less_than_equal
{"id":1,"name":"abc","age":67}
list[Model]例子:
class Item(BaseModel):id: intname: Annotated[str, Field(min_length=1, max_length=5)]def test():items_type = TypeAdapter(list[Item])obj = items_type.validate_python([{'id': 1, 'name': 'abc'}, {'id': 2, 'name': 'bcd'}])print(items_type.dump_json(obj).decode('utf-8'))if __name__ == '__main__':test()
输出:
[{"id":1,"name":"abc"},{"id":2,"name":"bcd"}]
validate_call 校验
允许在函数的使用类型注解参数上进行数据解析与校验.参数类型定义,可以和Model字段定义遵循一样的语法,可以结合annotated pattern 以及 pydantic Field等.举个完整的例子:
@validate_call(config=ConfigDict(strict=True), validate_return=True)
def test_val(a: int, b: str, c: Annotated[int, Ge(10)], d: Annotated[int, Field(le=5)]) -> str:return ' '.join([str(a), b, str(c), str(d)])def test():try:test_val('1', 'abc', 11, 6)except ValidationError as e:print(e.errors())if __name__ == '__main__':test()
输出:
[{'type': 'int_type', 'loc': (0,), 'msg': 'Input should be a valid integer', 'input': '1', 'url': 'https://errors.pydantic.dev/2.11/v/int_type'}, {'type': 'less_than_equal', 'loc': (3,), 'msg': 'Input should be less than or equal to 5', 'input': 6, 'ctx': {'le': 5}, 'url': 'https://errors.pydantic.dev/2.11/v/less_than_equal'}]
实战案例
这里举例pydantic在LLM应用开发中的两个例子.第一个是利用pydantic生成function schema,调用LLM时一并传递;第二个场景则是指定json schema,让LLM产生满足json schema定义的结构化的结果.
llm function schema
假设需要使用llm的function call功能,在不调用现有MCP server前提下本地实现了一个天气查询的函数,需要提供给LLM让其决定是否调用函数.一般的实现是定义好函数之后,额外去编写需要提供给LLM的function schema,比如:
tools = [{"type": "function","function": {"name": "get_weather","description": "Get weather of a location, the user should supply a location first.","parameters": {"type": "object","properties": {"location": {"type": "string","description": "The city and state, e.g. San Francisco, CA",}},"required": ["location"]},}},
]
这样函数如果进行了修改还需要同步修改这个function schema,这种实现比较不优雅,但是可以借助pydantic以及装饰器语法实现自动为编写的函数添加function schema.核心代码如下:
from pydantic import TypeAdapter, Field
from typing import Annotated, get_type_hints
import functools, inspect, jsondef function_schema(func):@functools.wraps(func)def wrapper(*args, **kwargs):return func(*args, **kwargs)func_schema_info = {"type": "function"}func_info = {}func_info['name'] = func.__name__if func.__doc__:func_info['description'] = func.__doc__func_params = {}signs = inspect.signature(func)hints = get_type_hints(func, include_extras=True)model_params = {}required_params = []for name, param in signs.parameters.items():ann = hints.get(name)if not ann:continue# required or notif param.default == inspect._empty:required_params.append(name)model_params[name] = annif model_params:func_params['type'] = 'object'if required_params:func_params['required'] = required_paramsproperties = {}for k, v in model_params.items():properties[k] = TypeAdapter(v).json_schema()func_params['propertis'] = propertiesif func_params:func_info['parameters'] = func_paramsfunc_schema_info['function'] = func_infowrapper.__func_schema__ = func_schema_inforeturn wrapper@function_schema
async def get_weather(location: Annotated[str, Field(description='The city and state, e.g. San Francisco, CA')]) -> str:'''Get weather of a location, the user should supply a location first.'''return '36℃'if __name__ == '__main__':res = get_weather.__func_schema__print(json.dumps(res, indent=2))
输出:
{"type": "function","function": {"name": "get_weather","description": "Get weather of a location, the user should supply a location first.","parameters": {"type": "object","required": ["location"],"propertis": {"location": {"description": "The city and state, e.g. San Francisco, CA","type": "string"}}}}
}
只要将funtion_schema装饰在函数上,同时参数显示提供类型注解以及参数的description,装饰器会实现解析参数并为函数生成function schema 添加到函数的__func__schema__魔术属性之上,完整的结合openai的调用的案例如下:
from pydantic import TypeAdapter, Field
from typing import Annotated, get_type_hints
import functools, inspect, json, asyncio
from openai import AsyncOpenAIdef function_schema(func):@functools.wraps(func)def wrapper(*args, **kwargs):return func(*args, **kwargs)func_schema_info = {"type": "function"}func_info = {}func_info['name'] = func.__name__if func.__doc__:func_info['description'] = func.__doc__func_params = {}signs = inspect.signature(func)hints = get_type_hints(func, include_extras=True)model_params = {}required_params = []for name, param in signs.parameters.items():ann = hints.get(name)if not ann:continue# required or notif param.default == inspect._empty:required_params.append(name)model_params[name] = annif model_params:func_params['type'] = 'object'if required_params:func_params['required'] = required_paramsproperties = {}for k, v in model_params.items():properties[k] = TypeAdapter(v).json_schema()func_params['propertis'] = propertiesif func_params:func_info['parameters'] = func_paramsfunc_schema_info['function'] = func_infowrapper.__func_schema__ = func_schema_inforeturn wrapper@function_schema
async def get_weather(location: Annotated[str, Field(description='城市的名字')]) -> str:'''根据输入的城市获取对应城市的温度信息'''cities_weather = {'北京': '13℃','武汉': '11℃','海口': '24℃'}return cities_weather.get(location, 'no such city data')client = AsyncOpenAI()sys_temp = '''你需要根据用户输入的需求(input)推荐旅游城市
'''
user_temp = 'input: {input_data}'async def send_messags(messages, model='deepseek-chat', tools=None):response = await client.chat.completions.create(messages=messages, model='deepseek-chat', tools=tools)return response.choices[0].messageasync def demo():input_data = '今年冬季我想去国内城市旅行,但是想去天气比较温暖的城市,我想在北京,上海,武汉,海口这几个城市选择,请推荐一下'messages = [{'role': 'system', 'content': sys_temp},{'role': 'user', 'content': user_temp.format(input_data=input_data)}]tools=[get_weather.__func_schema__]message = await send_messags(messages=messages, tools=tools)tool_messages = []messages.append(message)for tool_call in message.tool_calls:tool_id = tool_call.idtool_type = tool_call.typeif tool_type == 'function':func = tool_call.functionfunc_name = func.namefunc_args = json.loads(func.arguments)if func_name == 'get_weather':func_res = await get_weather(**func_args)tool_messages.append({'role': 'tool', 'tool_call_id': tool_id, 'content': func_res})if tool_messages:messages += tool_messagesres = await send_messags(messages=messages, tools=tools)print(res.content)if __name__ == '__main__':asyncio.run(demo())
输出:
根据查询到的天气数据,我为您分析这几个城市的冬季温暖程度:**城市温度对比:**
- **海口**:24℃ - 最温暖
- **北京**:13℃
- **武汉**:11℃
- **上海**:数据暂缺(但通常冬季温度与武汉相近)**推荐建议:**🏆 **强烈推荐:海口**
- 温度24℃,是这几个城市中最温暖的
- 冬季气候宜人,适合避寒旅行
- 作为热带滨海城市,冬季依然可以享受阳光沙滩**其他城市分析:**
- **北京**:13℃,相对较冷,需要穿厚衣服
- **武汉**:11℃,冬季湿冷,体感温度可能更低
- **上海**:预计温度与武汉相近,冬季也比较寒冷如果您想要在冬季享受温暖的天气,**海口**是您的最佳选择!它不仅有温暖的天气,还有美丽的海滩和热带风光,非常适合冬季避寒旅行。
llm 结构化输出以及解析
现在大模型都支持结构化输出(配置输出response format是json),可以使用pydantic去构建输出结果的模型,然后生成模型的json schema 添加到prompt中,同时也可以对LLM输出json字符串使用validate相关函数进行校验.举个例子,下面是利用LLM从非结构化文本中提取结构化信息并且解析成python对象的场景:
from openai import AsyncClient
from pydantic import BaseModel, Field, PositiveInt, TypeAdapter
from typing import Annotated
import asyncio, jsonclient = AsyncClient()async def send_messages(messages, model='deepseek-chat', tools=None):response = await client.chat.completions.create(messages=messages, model=model, tools=tools, response_format={'type': 'json_object'})return response.choices[0].messageclass Film(BaseModel):title: stryear: PositiveIntdirector: Annotated[str, Field(min_length=1)]summary: Annotated[str, Field(min_length=1)]# 注意template 中json sample引号要双引号防止转义
sys_temp = """你需对用户的输入的原始文本解析成json格式数据,解析出的结果需要遵循如下的json schema:提供一些输入输出的样例作为参考{json_schema}输入原始文本样例:输出json格式数据样例:{{"title": "肖申克的救赎","year": 1994,"director": "弗兰克·德拉邦特","summary": "银行家安迪被冤枉杀害妻子及其情人,被判终身监禁。在肖申克监狱中,他凭借智慧与毅力,最终实现自我救赎并成功越狱。"}}"""async def demo():output_schema = json.dumps(Film.model_json_schema(), indent=2)user_input = '''肖申克的救赎,1994,弗兰克·德拉邦特,银行家安迪被冤枉杀害妻子及其情人,被判终身监禁。在肖申克监狱中,他凭借智慧与毅力,最终实现自我救赎并成功越狱Inception,2010,Christopher Nolan,Dom Cobb is a skilled thief who steals secrets from within the subconscious during dream states. He's offered a chance to have his criminal record erased if he can successfully perform 'inception'—planting an idea in someone's mind.寄生虫,2019,봉준호,一个失业的底层家庭通过各种手段渗透进一个富裕家庭的生活,阶级矛盾逐渐激化,最终酿成悲剧。影片融合黑色幽默与社会批判,荣获第92届奥斯卡最佳影片。Pulp Fiction,1994,Quentin Tarantino,The lives of two mob hitmen, a boxer, a gangster's wife, and a pair of diner bandits intertwine in four tales of violence and redemption.霸王别姬,1993,陈凯歌,跨越半个世纪的爱恨情仇,讲述两位京剧演员在时代洪流中的命运沉浮,被誉为华语电影的巅峰之作。'''messages = [{'role': 'system', 'content': sys_temp.format(json_schema=output_schema)},{'role': 'user', 'content': user_input}]message = await send_messages(messages)res_content = message.contentfilms = TypeAdapter(list[Film]).validate_json(res_content)for film in films:print(repr(film))if __name__ == '__main__':asyncio.run(demo())
输出结果:
Film(title='肖申克的救赎', year=1994, director='弗兰克·德拉邦特', summary='银行家安迪被冤枉杀害妻子及其情人,被判终身监禁。在肖申克监狱中,他凭借智慧与毅力,最终实现自我救赎并成功越狱。')
Film(title='Inception', year=2010, director='Christopher Nolan', summary="Dom Cobb is a skilled thief who steals secrets from within the subconscious during dream states. He's offered a chance to have his criminal record erased if he can successfully perform 'inception'—planting an idea in someone's mind.")
Film(title='寄生虫', year=2019, director='봉준호', summary='一个失业的底层家庭通过各种手段渗透进一个富裕家庭的生活,阶级矛盾逐渐激化,最终酿成悲剧。影片融合黑色幽默与社会批判,荣获第92届奥斯卡最佳影片。')
Film(title='Pulp Fiction', year=1994, director='Quentin Tarantino', summary="The lives of two mob hitmen, a boxer, a gangster's wife, and a pair of diner bandits intertwine in four tales of violence and redemption.")
Film(title='霸王别姬', year=1993, director='陈凯歌', summary='跨越半个世纪的爱恨情仇,讲述两位京剧演员在时代洪流中的命运沉浮,被誉为华语电影的巅峰之作。')
小结
本文讲解的pydantic用法基本上能涵盖绝大多数的开发场景,如果后续pydantic有较大的版本升级,本文内容也会同步升级.
