WheatField
WheatField

使用 Pydantic 验证数据

November 28, 20241350 words, 7 min read
Authors

Annotated Validators

也即是 Annotated[T, BeforeValidator(fn)] 这种形式,常用的几种 Validator 有:

  • BeforeValidator
  • AfterValidator
  • WrapValidator
  • PlainValidator

BeforeValidator 和 AfterValidator 都是接收一个函数,这个函数可以包含验证逻辑(e.g., 密码长度是否合适)及修改逻辑(e.g., 添加前缀)。

WrapValidator 也是接受一个函数 ff(v,h)f \rightarrow f(v, h),但这个函数接收两个参数,一个是待验证的值,一个是 ValidatorFunctionWrapHandler,也即是其他 Validator 的执行函数。

这几个 Validator 的一个重要区别是执行顺序,这一点我研究了半天才有点头绪。

首先,结合 Annotated 的定义可以知道,Annotated 的执行顺序是从外到内,先执行最外层的,然后是内层的,即:

A(T,V1,V2,,Vn)=A(A(T,V1,V2,,Vn1),Vn)==A(A(A(T,V1),V2),,Vn)\begin{aligned} \mathcal{A}(T, \mathcal{V}_1, \mathcal{V}_2, \cdots, \mathcal{V}_n) &= \mathcal{A}(\mathcal{A}(T, \mathcal{V}_1, \mathcal{V}_2, \cdots, \mathcal{V}_{n-1}), \mathcal{V}_n) \\ &= \cdots \\ &= \mathcal{A}(\mathcal{A}(\mathcal{A}(T, \mathcal{V}_1), \mathcal{V}_2), \cdots, \mathcal{V}_n) \end{aligned}

再一点,BeforeValidator 是在 Pydantic 进行解析及验证之前执行, 这里的验证指的是其他 Validator 的执行,比如另外若干个 BeforeValidator 或者 AfterValidator。

而 AfterValidator 是在 Pydantic 进行解析及验证之后执行, WrapValidator 则是根据 Annotated 的执行顺序,在轮到它时,它会先执行它的内在逻辑,再让位给下一位 Validator。这听起来有点拗口,但实际上也确实如此,后面我会举个例子说明。

综合上面的几点特性,可以推理出以下几点:

  1. 所有 BeforeValidator 一定是先于所有 AfterValidator 执行的。
  2. 多个 AfterValidator 的执行顺序是先进先出(FIFO, first in first out),即先声明的先执行。这是因为有多个 AfterValidator 时,e.g., A(T,V1,V2)\mathcal{A}(T, \mathcal{V}_1, \mathcal{V}_2) 等价于 A(A(T,V1),V2)\mathcal{A}(\mathcal{A}(T, \mathcal{V}_1), \mathcal{V}_2),而 V2\mathcal{V}_2 会等待 A(T,V1)\mathcal{A}(T, \mathcal{V}_1) 执行完毕之后再执行。类似地,
  3. 多个 BeforeValidator 的执行顺序是后进先出(LIFO, last in first out),即后声明的先执行。

例子

只有 BeforeValidator

Annotated[int, BeforeValidator(f1), BeforeValidator(f2)]

根据 BeforeValidator 的 LIFO 特性,验证顺序是:f2 -> f1。

Before, After 混合使用

Annotated[int,
    BeforeValidator(f1),
    AfterValidator(f2),
    BeforeValidator(f3),
    BeforeValidator(f4),
    AfterValidator(f5),
]

根据以上推理 1,验证顺序是:f4 -> f3 -> f1 -> f2 -> f5。

WrapValidator

上面提到,WrapValidator 接收一个函数 ff(v,h)f \rightarrow f(v, h),这个函数接收一个值及一个 ValidatorFunctionWrapHandler。它可以封装自己的验证逻辑,并调用下一个 Validator 的执行函数。

执行过程中,它会先执行自己的前半部分逻辑,然后调用下一个 Validator 的执行函数,最后执行剩下的逻辑。举个例子:

def validate_length(v: str, h: ValidatorFunctionWrapHandler):
    print(f"V1 -- pre")
    if len(v) < 3:
        raise ValueError("太短了")
    x = h(v)
    print(f"V1 -- post, {x}")
    return x

def add_prefix(v: int, h: ValidatorFunctionWrapHandler):
    print(f"A1 -- pre")
    v = f"prefix-{v}"
    x = h(v)
    print(f"A1 -- post, {x}")
    return x

Xstr = Annotated[str,
                 WrapValidator(add_prefix),
                 WrapValidator(validate_length)                 ]

class X(BaseModel):
    x: Xstr
print(X(x="abc").x)

以上代码的输出是:

V1 -- pre
A1 -- pre
A1 -- post, prefix-abc
V1 -- post, prefix-abc
prefix-abc

现在看一下为什么是这个顺序。

  • 因为 Annotated 从外向内执行,因此首先执行 WrapValidator(validate_length),所以会先打印 V1 -- pre;
  • 打印完就遇到 x = h(v),也就是说它要让位给下一个验证器进行验证,这里下一个验证器是 WrapValidator(add_prefix),所以会执行 add_prefix 并打印 A1 -- pre;
  • add_prefix 执行完会之前打印 A1 -- post, prefix-abc,然后返回 prefix-abc,并回到 validate_length 继续执行剩下的逻辑;
  • validate_length 拿到新的 x 并打印 V1 -- post, prefix-abc,最终返回 prefix-abc

Before, After, Wrap 混打

再看一下复杂亿点点的例子:

def f(label: str):
    def validator(v: Any, info: ValidationInfo) -> Any:
        print(f"{label}")
        return v
    return validator

def g(label: str):
    def validator(
        v: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
    ) -> Any:
        print(f"{label}: pre")
        result = handler(v)
        print(f"{label}: post")
        return result

    return validator


class A(BaseModel):
    x: Annotated[
        str,
        AfterValidator(f('after-1')),
        WrapValidator(g('wrap-1')),
        BeforeValidator(f('before-1')),
        WrapValidator(g('wrap-2')),
        BeforeValidator(f('before-2')),
        AfterValidator(f('after-2')),
        AfterValidator(f('after-3')),
    ]

A.model_validate({'x': 'abc'})

输出结果:

before-2
wrap-2: pre
before-1
wrap-1: pre
after-1
wrap-1: post
wrap-2: post
after-2
after-3

分析一下:

  • 从外向内,看到 Before 就先执行 Before(LIFO),如果有 Wrap,接着执行 Wrap 的内部逻辑(LIFO),e.g., wrap-2: pre -> wrap-1: pre,但不执行 handler
  • Before 执行完,看看内部还有没有其他 Validator,如果有,接着执行(FIFO),e.g., after-1;
  • 如果内部没有其他 Validator,Wrap 在 handler 执行完后,执行自己后续的逻辑(LIFO),e.g., wrap-1: post -> wrap-2: post
  • 最后执行剩下的 After(FIFO),e.g., after-2 -> after-3

所以可以看到 WrapValidator 非常灵活(复杂),本质上就是一个装饰器,可以封装自己的验证逻辑,并调用下一个 Validator 的执行函数。结合 Before, After validator,可以实现很多复杂的功能。

当然如果可以,我还是倾向于仅使用 Before, After validator。参照 Python 哲学, keep it simple and stupid。能嵌套如示例这么多 Validator 的场景,我还真没有遇到过。

实战 Q & A

验证一个值的属性有哪些方式?

比如验证一个值是否是质数。

def is_prime(n: int) -> bool:
    ...

nums = [17, "23", 15, 4, "31"]
for num in nums:
    try:
        n = do_validate(num)
        print(n, type(n))
    except ValueError as e:
        print(e)

1. 使用 TypeAdapter

from pydantic import TypeAdapter, BeforeValidator, Annotated
PrimeInt = Annotated[int, BeforeValidator(validate_prime)]
prime_adapter = TypeAdapter(PrimeInt)

2. 使用 Annotated + BaseModel

from pydantic import BaseModel, Field
PrimeInt = Annotated[int, BeforeValidator(validate_prime)]

class PrimeModel(BaseModel):
    num: PrimeInt

这种方式跟上面的 TypeAdapter 类似,区别在于 TypeAdapter 更轻量,创建一个实例,可重复使用,而 BaseModel 需要使用都需要创建。这一点在之前的博客 Pydantic TypeAdapter 中也有提到。

3. 使用 @field_validator

from pydantic import BaseModel, field_validator
class PrimeModel(BaseModel):
    num: int = Field()

    @field_validator('num')
    def validate_num(cls, v):
        if not is_prime(v):
            raise ValueError(f'{v} is not a prime number')
        return int(v)

参考资料