WheatField
WheatField

Pydantic 序列化的一些小技巧

November 10, 20242221 words, 12 min read
Authors

《Pydantic 初探-有哪些常见写法》 中,我们介绍了 Pydantic 的常见写法。 这篇文章我们来看看 Pydantic 的序列化。

说到序列化我们主要指两个方面:一是把模型转换成 dict,二是把模型转换成 json 字符串。常用的场景是在格式化输出日志、打印模型信息时,通过这两个方法来实现自定义输出。

常规操作

model_dump 将模型转换成 dict

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

user = User(name="Tom", age=20)
print(user.model_dump())

在 2.0 之前的版本中,把模型转成字典的 API 是 model.dict() 或者 dict(model)。 从 2.x 开始推荐使用 model_dump 这种写法,主要有几点原因:

  1. model_dump 对比 dict 的第一个好处是:速度更快
  2. model_dump 是可以通过序列化规则(比如下面聊到的 model serializer)重写返回类型的,这时再继续使用 dict 就有点误导。

关于二者速度对比,我在线下的基准测试中,把下面这个对象用两个方法分别序列化 100000 次, 重复 5 轮,对比测试结果如下,可以看到 model_dumpdict 快 0.76 倍。

user= User(name="Tom", age=20, email="[email protected]", is_active=True, scores=[85, 90, 95])
MethodMean Time (s)Std Dev (s)Avg per op (μs)Speedup
.model_dump()0.08480.00060.851.76x
.dict()0.14950.00081.491.00x
dict(user)0.14980.00111.501.00x

直接当 dict 使用

这点没什么新奇的,因为实现了 __iter__ 方法,Pydantic 模型可以直接当 dict 使用。

for name, value in user:
    print(f'{name}: {value}')

有选择的序列化字段

在进行序列化操作时,有时我们希望只序列化某些字段。比如在日志打印过程中,不希望把敏感信息也打印出来。 那通过 includeexclude 参数,可以有选择的序列化字段。

print(user.model_dump(include={'name', 'age'}))
# 只序列化除 name 和 age 以外的字段
print(user.model_dump(exclude={'name', 'age'}))
# 序列化除 name 和 age 以外的字段

除了这种显式传递 includeexclude 参数,也可以在声明模型时,通过 Fieldexclude 参数来指定。

class User(BaseModel):
    name: str = Field(exclude=True)
    age: int

后面再序列化时,name 字段就不会被序列化。需要注意的是:Field(exclude=True) 参数优先级高于 include 和 exclude 参数,也就是即使显式传递了 include、exclude 参数,已经被 Field(exclude=True) 的字段也不会被序列化。

结合上下文序列化

我们知道,在序列化时可以通过 field_serializer 装饰器来改变序列化行为,比如给某个字段添加一些处理逻辑。 如果这个处理过程中,需要使用到一些“额外”信息,那如何把这些信息传递给序列化函数呢?

一种方法是通过一个全局变量来传递,在序列过程中,从全局变量中获取一些信息,再根据这个信息更改字段的序列化行为。

Pydantic 提供了一种更方便的方法,那就是通过 context 参数来传递。


class Model(BaseModel):
    text: str
    ip: str

    @field_serializer('ip')
    def hide_some_ip(self, v: str, info: SerializationInfo):
        hide_list: List[str] = info.context.get('hide_list', [])
        if hide_list:
            if v in hide_list:
                return 'hidden'
        return v


model = Model.model_construct(
    **{'text': 'This is an example document', 'ip': '192.168.1.1'})
print(model.model_dump(context={'hide_list': ['192.168.1.1']}))

除了 context 之外,info: SerializationInfo 这个类型包含的内容还挺多的,比如 include、exclude、mode 等。

class SerializationInfo(Protocol):
    @property
    def include(self) -> IncExCall: ...

    @property
    def exclude(self) -> IncExCall: ...

    @property
    def context(self) -> Any | None:

    @property
    def mode(self) -> str: ...

    @property
    def by_alias(self) -> bool: ...

    @property
    def exclude_unset(self) -> bool: ...

    @property
    def exclude_defaults(self) -> bool: ...

    @property
    def exclude_none(self) -> bool: ...

    @property
    def serialize_as_any(self) -> bool: ...

    def round_trip(self) -> bool: ...

    def mode_is_json(self) -> bool: ...

model_dump_json 将模型转换成 json 字符串

print(user.model_dump_json())

整体用法跟上面 model_dump 基本一致,唯一的区别是返回的是 json 字符串。之前的 API 是 user.json(),现在推荐使用 model_dump_json

自定义序列化器

上面提取在序列化时,可以筛选过滤一些敏感字段,或者传递一些上下文信息。 而这些都还只是基本操作,Pydandic 还能做点更复杂的操作,比如在序列化时,更改一些字段的值,或者更改整个模型对应的结构信息。

字段序列化器 field_serializer

结合 field_serializer 装饰器可以对特定字段进行自定义序列化,这里的自定义一般指输出格式。比如模糊化敏感信息,或者将 datetime 字段序列化为 ISO 格式,再或者对值进行计算、格式化等。

class Model(BaseModel):
    text: str
    password: SecretStr
    dt: datetime

    @field_serializer('dt')
    def serialize_dt(self, v: datetime, info: SerializationInfo):
        return v.isoformat()

    @field_serializer('password')
    def serialize_password(self, v: SecretStr, info: SerializationInfo):
        return '**********'

模型序列化器 model_serializer

当然,如果想更改整个模型的默认序列化行为,可以使用 model_serializer 装饰器。

class Model(BaseModel):
    text: str
    dt: datetime

    @model_serializer
    def set_model(self)->typing.Dict[str, Any]:
        return {'text': self.text[:42], 'dt': self.dt.isoformat()}

相当于在 model_dump 时,调用 set_model 函数,自己可以任意定制序列化行为:隐藏一些字段(有点类似上面提到的 exclude)、添加一些字段、修改一些字段的值。

@model_serializer 还有一个功能就是:更改序列化的返回类型。这一点比较重要,默认情况下都是字典,但如果需要列表、集合甚至 bytes,也是可以操作的,只需要在 return 时,返回对应的类型即可。需要注意的是,这不会影响 print(model) 的输出,打印结果仍然是字典。

这个功能我觉得挺鸡肋的,因为如果想得到特定类型的输出,我大可以自定义函数,比如 to_list()to_bytes() 等,完全没有必要对一个大家都理所当然的认为该输出字典的函数进行返回类型上的炫技式改造,即不优雅也不方便维护。但凡后面出了点 bug,维护者首先想到的肯定不是:“哦,应该是 core/libs/model/user.py 中的 user 的 model_dump 返回类型非字典导致的”。

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

    @model_serializer
    def set_model(self)->typing.List[str]:
        return [self.name, self.age]

函数式序列化器

同样,也是自定义序列化的一种方式,区别在于什么时机下开始改造。主要有两个函数式序列化器:PlainSerializerWrapSerializer

从定义可以看到,PlainSerializer 接收一个函数(序列化函数),一个返回值类型 return_type,及一个调用场景 when_used。它做的事情是:当 when_used 参数指定的场景下,调用序列化函数,并用 return_type 参数指定返回值类型。

PlainSerializer(
    func: SerializerFunction,
    return_type: Any = PydanticUndefined,
    when_used: WhenUsed = "always",
)

when_used 这个参数有几个可选值:

  • always:总是使用序列化函数
  • json:只在 JSON 序列化时使用序列化函数
  • json-unless-none:只在 JSON 序列化时使用序列化函数,除非字段的值是 None
  • unless-none:不使用序列化函数,除非字段的值是 None

PlainSerializerfield_serializer 的功能类似,都可以改变序列化时一个字段的输出形式。 不同的是,PlainSerializer需要结合 Annotated 一起使用来定义序列化规则。 下例中,在通过 model_dump_json 序列化时,将名字序列化为 lastname 在前,firstname 在后的格式。

from typing import Annotated
from pydantic import BaseModel
from pydantic.functional_serializers import PlainSerializer

LastNameFirst = Annotated[
    str, PlainSerializer(lambda x: ' '.join(
        x.split(' ')[::-1]), return_type=str, when_used='json')
]

class User(BaseModel):
    name: LastNameFirst

print(User(name='Tom Li').model_dump())
# > {'name': 'Tom Li'}

print(User(name='Tom Li').model_dump(mode='json'))
# > {'name': 'Li Tom'}

WrapSerializer 的用法也很类似,接收原始输入、一个序列化函数和序列化信息,返回的值即为最终序列化结果。

from typing import Any
import httpx
from typing import Annotated
from pydantic import BaseModel, SerializerFunctionWrapHandler
from pydantic.functional_serializers import WrapSerializer

def get_city(ip: str) -> str:
    with httpx.Client() as client:
        resp = client.get(f'https://ipapi.co/{ip}/json/')
        return resp.json()['city']

def withcity(ip: Any, handler, info) -> str:
    city = get_city(ip)
    return f'{ip} / {city}'

GeoLocation = Annotated[str, WrapSerializer(withcity)]
class MyModel(BaseModel):
    ip: GeoLocation

print(MyModel(ip='223.17.172.124').model_dump())
# > {'ip': '223.17.172.124 / Kwun Tong'}

withcity 中有三个参数,这三个参数的含义是:

  • value: 即原始输入值,在这个例子中是 EventDatetime 实例
  • handler: 这是一个 SerializerFunctionWrapHandler 类型的函数,它封装了默认的序列化逻辑。我们可以调用 handler(value, info) 来执行默认的序列化过程。在 wrap 模式下,这让我们可以:
    • 先使用默认序列化
    • 然后对结果进行自定义修改
  • info: 这是一个 SerializationInfo 对象,包含序列化的上下文信息,具体参考上面 SerializationInfo 的定义。

model_copy 复制模型

model_copy 可以复制模型,并有选择的更新字段,在复制过程中,也存在 deep copy 与 shallow copy 的区别,运行中实际上调用的就是 copy 模块的 copydeepcopy 函数。

cp = user.model_copy(update={'name': 'Jerry'}, deep=True)
# 即深拷贝并更新 name 字段
print(cp)

小节 & 参考

序列化操作是开发中比较常见的操作,Pydantic 提供了多种方式来实现自定义序列化,本文介绍的这些方法基本可以覆盖大部分的使用场景。