Pydantic 自定义验证器实战基于 Annotated 元数据与嵌套模型验证的完整指南【免费下载链接】pydanticData validation using Python type hints项目地址: https://gitcode.com/GitHub_Trending/py/pydantic本文是 Pydantic 自定义验证器的高级实战指南聚焦两类高频场景通过Annotated类型元数据与__get_pydantic_core_schema__协议构造可复用的自定义验证器以datetime时区与 UTC 偏移约束为例以及跨模型/跨字段的嵌套验证外层model_validator与基于 validation context 的field_validator两种方案。文中所有示例均可在当前仓库直接运行读者学完后即可在自己的 Pydantic 模型中封装带参数的验证器、利用 wrap 验证器在标准验证前后插入逻辑并通过 context 在嵌套模型间共享数据。本文示例均改编自 Pydantic 的 issues 与社区讨论旨在展示 Pydantic 验证系统的灵活性与扩展能力。前置知识Pydantic 的自定义验证体系在深入示例前先理清 Pydantic 自定义验证涉及的两个核心扩展点__get_pydantic_core_schema__协议在类型或包装类型的元数据对象上实现该方法即可在 Pydantic 生成 Core Schema 时插入自定义逻辑。GetCoreSchemaHandler的完整接口定义位于 pydantic/annotated_handlers.py调用handler(source_type)会委托给下一层 schema 生成器得到该类型标准的CoreSchemahandler.generate_schema()则可生成与当前上下文无关的 schema。验证器函数形态Pydantic 验证器分为before验证前、after验证后、wrap包裹可同时在前与在后执行逻辑三类。其中 wrap 验证器通过pydantic_core的core_schema.no_info_wrap_validator_function()/with_info_wrap_validator_function()等工厂函数构建其类型定义位于 pydantic-core/python/pydantic_core/core_schema.py属于function-wrap类型的WrapValidatorFunctionSchema。此外Pydantic 官方在 pydantic/functional_validators.py 中提供的AfterValidator、BeforeValidator、WrapValidator等元数据类本质上也是通过实现__get_pydantic_core_schema__把用户函数翻译成上述core_schema工厂调用。理解了这一层就能自行封装任意形态的自定义验证器。自定义 datetime 验证器通过 Annotated 元数据附加时区约束本示例构造一个挂载在Annotated[dt.datetime, ...]上的自定义验证器MyDatetimeValidator它约束datetime对象必须满足给定的时区timezone要求。该验证器支持以字符串形式指定时区如America/Los_Angeles若datetime的时区不符合约束则抛出验证错误使用wrap 验证器以便在 Pydantic 对datetime的标准验证前后都能执行自定义逻辑。import datetime as dt from dataclasses import dataclass from pprint import pprint from typing import Annotated, Any import pytz from pydantic_core import CoreSchema, core_schema from pydantic import ( GetCoreSchemaHandler, PydanticUserError, TypeAdapter, ValidationError, ValidatorFunctionWrapHandler, ) dataclass(frozenTrue) class MyDatetimeValidator: tz_constraint: str | None None def tz_constraint_validator( self, value: dt.datetime, handler: ValidatorFunctionWrapHandler, # (1)! ): Validate tz_constraint and tz_info. # handle naive datetimes if self.tz_constraint is None: assert ( value.tzinfo is None ), tz_constraint is None, but provided value is tz-aware. return handler(value) # validate tz_constraint and tz-aware tzinfo if self.tz_constraint not in pytz.all_timezones: raise PydanticUserError( fInvalid tz_constraint: {self.tz_constraint}, codeunevaluable-type-annotation, ) result handler(value) # (2)! assert self.tz_constraint str( result.tzinfo ), fInvalid tzinfo: {str(result.tzinfo)}, expected: {self.tz_constraint} return result def __get_pydantic_core_schema__( self, source_type: Any, handler: GetCoreSchemaHandler, ) - CoreSchema: return core_schema.no_info_wrap_validator_function( self.tz_constraint_validator, handler(source_type), ) LA America/Los_Angeles ta TypeAdapter(Annotated[dt.datetime, MyDatetimeValidator(LA)]) print( ta.validate_python(dt.datetime(2023, 1, 1, 0, 0, tzinfopytz.timezone(LA))) ) # 2023-01-01 00:00:00-07:53 LONDON Europe/London try: ta.validate_python( dt.datetime(2023, 1, 1, 0, 0, tzinfopytz.timezone(LONDON)) ) except ValidationError as ve: pprint(ve.errors(), width100) [{ctx: {error: AssertionError(Invalid tzinfo: Europe/London, expected: America/Los_Angeles)}, input: datetime.datetime(2023, 1, 1, 0, 0, tzinfoDstTzInfo Europe/London LMT-1 day, 23:59:00 STD), loc: (), msg: Assertion failed, Invalid tzinfo: Europe/London, expected: America/Los_Angeles, type: assertion_error, url: https://errors.pydantic.dev/2/v/assertion_error}] handler即 wrap 验证器接收的内部验证回调调用它会对输入执行标准 Pydantic 验证此处再次调用handler(value)以标准验证得到的result作为后续断言对象。代码拆解与底层原理handler回调的类型ValidatorFunctionWrapHandler定义在 pydantic-core/python/pydantic_core/core_schema.py签名是__call__(input_value, outer_locationNone)。它正是no_info_wrap_validator_function所描述的洋葱式中间件模型——与许多 Web 框架的中间件实现思路一致外层函数可以在调用handler之前执行前置逻辑在handler返回之后执行后置逻辑。no_info的含义no_info_wrap_validator_function构建的是function{type: no-info, ...}的 schema即验证函数不接收ValidationInfo参数。若需要在验证器中访问 context、data、field_name 等信息应改用with_info_wrap_validator_function其函数签名为(input_value, validator, info)。handler(source_type)的委托在__get_pydantic_core_schema__中handler(source_type)返回dt.datetime的内置 Core Schema再将其作为schema参数传给 wrap 工厂函数——这保证了自定义验证逻辑永远包裹在标准验证之外。TypeAdapter的用途TypeAdapter允许在不定义BaseModel的情况下直接对任意类型此处是Annotated[dt.datetime, MyDatetimeValidator(LA)]进行验证与序列化是测试与复用自定义验证器的轻量入口。异常类型的选择tz_constraint本身不合法时抛出PydanticUserError配置/使用层面的错误而输入数据不满足约束时则通过assert触发ValidationError——两者语义区分明确值得借鉴。扩展强制 UTC 偏移量边界类似的思路可以用于约束datetime的UTC 偏移量。假设有lower_bound与upper_bound单位为小时自定义验证器保证datetime的 UTC 偏移落在闭区间内import datetime as dt from dataclasses import dataclass from pprint import pprint from typing import Annotated, Any import pytz from pydantic_core import CoreSchema, core_schema from pydantic import ( GetCoreSchemaHandler, TypeAdapter, ValidationError, ValidatorFunctionWrapHandler, ) dataclass(frozenTrue) class MyDatetimeValidator: lower_bound: int upper_bound: int def validate_tz_bounds( self, value: dt.datetime, handler: ValidatorFunctionWrapHandler ): Validate and test bounds assert value.utcoffset() is not None, UTC offset must exist assert self.lower_bound self.upper_bound, Invalid bounds result handler(value) hours_offset value.utcoffset().total_seconds() / 3600 assert ( self.lower_bound hours_offset self.upper_bound ), Value out of bounds return result def __get_pydantic_core_schema__( self, source_type: Any, handler: GetCoreSchemaHandler, ) - CoreSchema: return core_schema.no_info_wrap_validator_function( self.validate_tz_bounds, handler(source_type), ) LA America/Los_Angeles # UTC-7 or UTC-8 ta TypeAdapter(Annotated[dt.datetime, MyDatetimeValidator(-10, -5)]) print( ta.validate_python(dt.datetime(2023, 1, 1, 0, 0, tzinfopytz.timezone(LA))) ) # 2023-01-01 00:00:00-07:53 LONDON Europe/London try: print( ta.validate_python( dt.datetime(2023, 1, 1, 0, 0, tzinfopytz.timezone(LONDON)) ) ) except ValidationError as e: pprint(e.errors(), width100) [{ctx: {error: AssertionError(Value out of bounds)}, input: datetime.datetime(2023, 1, 1, 0, 0, tzinfoDstTzInfo Europe/London LMT-1 day, 23:59:00 STD), loc: (), msg: Assertion failed, Value out of bounds, type: assertion_error, url: https://errors.pydantic.dev/2/v/assertion_error}] 这段代码展示了 wrap 验证器三段式的典型结构前置断言在调用handler(value)之前检查输入的基本性质如utcoffset()必须存在、边界本身必须合法委托验证result handler(value)执行标准 Pydantic 验证字符串转datetime等解析工作由此完成后置断言对验证结果做业务约束检查lower_bound hours_offset upper_bound。注意assert失败会以ValidationErrortypeassertion_error的形式上报其结构化错误信息ctx.error、msg、loc、type、url可直接被上层工具消费。验证嵌套模型字段两种跨模型校验方案当某个字段的合法性依赖父模型中的数据时例如每个用户的密码不能出现在父模型定义的禁用密码列表中Pydantic 提供了两种典型实现路径。方案一在外层模型上使用 model_validator第一种做法是把校验逻辑放在外层模型的model_validator(modeafter)中——该验证器在所有字段验证完成后运行因此可以同时访问self.users与self.forbidden_passwordsfrom typing_extensions import Self from pydantic import BaseModel, ValidationError, model_validator class User(BaseModel): username: str password: str class Organization(BaseModel): forbidden_passwords: list[str] users: list[User] model_validator(modeafter) def validate_user_passwords(self) - Self: Check that user password is not in forbidden list. Raise a validation error if a forbidden password is encountered. for user in self.users: current_pw user.password if current_pw in self.forbidden_passwords: raise ValueError( fPassword {current_pw} is forbidden. Please choose another password for user {user.username}. ) return self data { forbidden_passwords: [123], users: [ {username: Spartacat, password: 123}, {username: Iceburgh, password: 87}, ], } try: org Organization(**data) except ValidationError as e: print(e) 1 validation error for Organization Value error, Password 123 is forbidden. Please choose another password for user Spartacat. [typevalue_error, input_value{forbidden_passwords: [...gh, password: 87]}, input_typedict] 该方案的优点逻辑集中、直观——所有跨字段/跨模型规则集中在一个方法中错误定位在模型层级loc指向Organization错误消息可以写得非常具体此处明确指出是哪个用户使用了哪个禁用密码。方案二在嵌套模型上使用 field_validator validation context第二种做法把校验逻辑下沉到嵌套模型User的field_validator中父模型的禁用密码列表则通过validation context验证上下文传入。这里的技巧是在父模型上再挂一个field_validator把forbidden_passwords字段的值写入 context供子模型读取。from pydantic import BaseModel, ValidationError, ValidationInfo, field_validator class User(BaseModel): username: str password: str field_validator(password, modeafter) classmethod def validate_user_passwords( cls, password: str, info: ValidationInfo ) - str: Check that user password is not in forbidden list. forbidden_passwords ( info.context.get(forbidden_passwords, []) if info.context else [] ) if password in forbidden_passwords: raise ValueError(fPassword {password} is forbidden.) return password class Organization(BaseModel): forbidden_passwords: list[str] users: list[User] field_validator(forbidden_passwords, modeafter) classmethod def add_context(cls, v: list[str], info: ValidationInfo) - list[str]: if info.context is not None: info.context.update({forbidden_passwords: v}) return v data { forbidden_passwords: [123], users: [ {username: Spartacat, password: 123}, {username: Iceburgh, password: 87}, ], } try: org Organization.model_validate(data, context{}) except ValidationError as e: print(e) 1 validation error for Organization users.0.password Value error, Password 123 is forbidden. [typevalue_error, input_value123, input_typestr] 该方案的优点校验逻辑与数据模型内聚——User知道自己的密码规则未来在其他模型中复用User时规则依然生效错误定位精确到具体字段路径users.0.password方便错误排查与前端提示。重要警告在验证器中修改 context 赋予了嵌套验证极强的能力但也容易写出难以调试的代码。请自行评估风险谨慎使用此模式关键注意事项context 缺失时的行为必须特别强调的是如果调用model_validate()时没有传入context参数那么验证器中的info.context将是None此时上述add_context不会把禁用密码列表写入 contextvalidate_user_passwords()也就不会执行密码校验。因此使用该模式时调用方必须显式传入context{}或至少是一个 dictorg Organization.model_validate(data, context{}) # 必须传入 context这一约束的根源在于context 由调用方在验证入口如model_validate(data, context...)注入具体行为可参见 docs/concepts/validators.md 中关于 validation context 的说明以及BaseModel.model_validate的签名见 pydantic/main.py 附近的__get_pydantic_core_schema__及main.py中模型验证入口。另外注意直接实例化模型Model(...)时目前无法传入 context官方文档给出的绕过方案是借助contextvars.ContextVar与自定义__init__同样详见 docs/concepts/validators.md。若想了解 context 在验证器中的完整读取方式如ValidationInfo.context与ValidationInfo.data的区别以及验证器执行顺序规则可阅读 docs/concepts/validators.md 与 pydantic/functional_validators.py 中的field_validator、model_validator实现。让自定义验证器的错误消息更有价值在上述示例中raise ValueError(...)中编写的消息值得用心打磨——它们不仅是开发调试时阅读的文本更是规则拒绝真实数据时用户看到的唯一线索。Pydantic 会把这些消息原样带入结构化错误输出如msg字段、ctx.error并传导给下游消费结构化错误的工具链例如 docs/errors/troubleshooting.md 中提到的 Logfire 对失败验证的解释就会包含自定义验证器的消息。好的错误消息应说明违反了什么规则、当前值是什么、期望值是什么参考第一个示例Invalid tzinfo: Europe/London, expected: America/Los_Angeles。小结本文围绕 docs/examples/custom_validators.md 的示例完整演示了两类自定义验证器的构建方法场景推荐工具底层机制关键文件对某类型附加带参数约束时区、偏移等Annotated 自定义元数据类 __get_pydantic_core_schema__wrap 验证器no_info_wrap_validator_functionpydantic/annotated_handlers.py、core_schema.py跨模型/跨字段校验外层model_validator(modeafter)模型级后置验证可访问全部字段pydantic/functional_validators.py子模型内校验 父数据注入field_validator validation contextcontext 由model_validate(context...)注入验证器中可变docs/concepts/validators.md进一步延伸若只需简单的前/后/包裹校验可优先使用官方提供的BeforeValidator/AfterValidator/WrapValidator元数据类见 pydantic/functional_validators.py它们与本文自定义元数据类走的是同一条__get_pydantic_core_schema__路径需要精确控制验证器顺序时可参考 docs/concepts/validators.md 中关于 before/wrap/after 执行顺序的说明。【免费下载链接】pydanticData validation using Python type hints项目地址: https://gitcode.com/GitHub_Trending/py/pydantic创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考