验证器

警告

“🚧 正在进行中” 此页面正在进行中。

此页面提供了在 Pydantic 中创建更复杂、自定义验证器的示例代码片段。

使用带有 `Annotated` 元数据的自定义验证器¶

在这个示例中，我们将构建一个自定义验证器，附加到一个 Annotated 类型上，以确保 datetime 对象遵守给定的时区约束。

自定义验证器支持时区的字符串指定，如果 datetime 对象没有正确的时区，它将引发错误。

我们在验证器中使用 __get_pydantic_core_schema__ 来定制已注释类型的模式（在这种情况下， datetime ），这使我们能够添加自定义验证逻辑。值得注意的是，我们使用 wrap 验证器函数，以便我们可以在默认的 pydantic 对 datetime 进行验证之前和之后执行操作。

import datetime as dt
from dataclasses import dataclass
from pprint import pprint
from typing import Any, Callable, Optional

import pytz
from pydantic_core import CoreSchema, core_schema
from typing_extensions import Annotated

from pydantic import (
    GetCoreSchemaHandler,
    PydanticUserError,
    TypeAdapter,
    ValidationError,
)


@dataclass(frozen=True)
class MyDatetimeValidator:
    tz_constraint: Optional[str] = None

    def tz_constraint_validator(
        self,
        value: dt.datetime,
        handler: Callable,  # (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(
                f'Invalid tz_constraint: {self.tz_constraint}',
                code='unevaluable-type-annotation',
            )
        result = handler(value)  # (2)!
        assert self.tz_constraint == str(
            result.tzinfo
        ), f'Invalid 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, tzinfo=pytz.timezone(LA)))
)
#> 2023-01-01 00:00:00-07:53

LONDON = 'Europe/London'
try:
    ta.validate_python(
        dt.datetime(2023, 1, 1, 0, 0, tzinfo=pytz.timezone(LONDON))
    )
except ValidationError as ve:
    pprint(ve.errors(), width=100)
    """
    [{'ctx': {'error': AssertionError('Invalid tzinfo: Europe/London, expected: America/Los_Angeles')},
    'input': datetime.datetime(2023, 1, 1, 0, 0, tzinfo=<DstTzInfo '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.8/v/assertion_error'}]
    """

handler 函数是我们用来验证输入的标准 pydantic 验证
我们在此包装验证器中调用 handler 函数，以使用标准 pydantic 验证来验证输入

我们也可以以类似的方式强制 UTC 偏移量约束。假设我们有一个 lower_bound 和一个 upper_bound ，我们可以创建一个自定义验证器来确保我们的 datetime 的 UTC 偏移量在我们定义的边界内是包含的：

import datetime as dt
from dataclasses import dataclass
from pprint import pprint
from typing import Any, Callable

import pytz
from pydantic_core import CoreSchema, core_schema
from typing_extensions import Annotated

from pydantic import GetCoreSchemaHandler, TypeAdapter, ValidationError


@dataclass(frozen=True)
class MyDatetimeValidator:
    lower_bound: int
    upper_bound: int

    def validate_tz_bounds(self, value: dt.datetime, handler: Callable):
        """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, tzinfo=pytz.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, tzinfo=pytz.timezone(LONDON))
        )
    )
except ValidationError as e:
    pprint(e.errors(), width=100)
    """
    [{'ctx': {'error': AssertionError('Value out of bounds')},
    'input': datetime.datetime(2023, 1, 1, 0, 0, tzinfo=<DstTzInfo '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.8/v/assertion_error'}]
    """

验证嵌套模型字段¶

在这里，我们展示了两种验证嵌套模型字段的方法，其中验证器利用了来自父模型的数据。

在这个例子中，我们构建了一个验证器，用于检查每个用户的密码是否不在父模型指定的禁止密码列表中。

有一种方法是在外部模型上放置一个自定义验证器：

from typing import List

from 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(mode='after')
    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(
                    f'Password {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. [type=value_error, input_value={'forbidden_passwords': [...gh', 'password': '87'}]}, input_type=dict]
    """

或者，可以在嵌套模型类（ User ）中使用自定义验证器，并通过验证上下文传递来自父模型的禁止密码数据。

警告

在验证器中更改上下文的能力为嵌套验证添加了很多功能，但也可能导致令人困惑或难以调试的代码。请自行承担使用此方法的风险！

from typing import List

from pydantic import BaseModel, ValidationError, ValidationInfo, field_validator

class User(BaseModel): username: str password: str

@field_validator('password', mode='after')
@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(f'Password {password} is forbidden.')
    return password

class Organization(BaseModel): forbidden_passwords: List[str] users: List[User]

@field_validator('forbidden_passwords', mode='after')
@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. [type=value_error, input_value='123', input_type=str] """

请注意，如果上下文属性未包含在 model_validate 中，那么 info.context 将是 None ，并且禁止密码列表将不会被添加到上述实现中的上下文。因此， validate_user_passwords 将不会执行所需的密码验证。

有关验证上下文的更多详细信息可以在这里找到。

本文总阅读量次

验证器

使用带有 Annotated 元数据的自定义验证器¶

验证嵌套模型字段¶

使用带有 `Annotated` 元数据的自定义验证器¶