Python Pydantic 序列化详解：从基础转换到高级定制

Pydantic 序列化基础与核心差异

Pydantic 模型转储为字典或 JSON 字符串的标准方法是调用 model_dump。该方法会递归处理所有嵌套的子模型，将其完整转化为字典结构。需注意，直接使用 dict(model) 也能将模型转为字典，但嵌套对象不会被递归解析，会导致数据结构扁平化丢失，因此在复杂模型中必须优先使用 model_dump。

实现自定义序列化逻辑

当默认序列化规则无法满足业务需求（如时间格式调整、敏感字段脱敏或结构重组）时，可通过装饰器介入序列化链。

第1步：通过 @field_serializer 实现字段级自定义序列化

目的：修改特定字段的输出格式或值。
入口：定义在 Pydantic BaseModel 子类内部。
动作：使用 @field_serializer('字段名') 装饰实例方法或静态方法。
重要参数：默认为 mode='plain'，完全跳过 Pydantic 内置逻辑；设置 mode='wrap' 可接入序列化链，允许在 Pydantic 处理前后添加前置或后置逻辑。方法签名根据模式不同分为四种：仅接收值和上下文信息，或额外接收 nxt: SerializerFunctionWrapHandler 处理器。
完成状态：模型调用 model_dump_json() 时，目标字段输出符合自定义函数的返回值。

from datetime import datetime, timedelta, timezone
from pydantic import BaseModel, ConfigDict, field_serializer
from pydantic_core.core_schema import FieldSerializationInfo, SerializerFunctionWrapHandler

class WithCustomEncoders(BaseModel):
    model_config = ConfigDict(ser_json_timedelta='iso8601')
    dt: datetime
    diff: timedelta
    diff2: timedelta

    @field_serializer('dt')
    def serialize_dt(self, dt: datetime, _info: FieldSerializationInfo):
        print(_info)
        return dt.timestamp()

    @field_serializer('diff')
    def ssse(self, diff: timedelta, info: FieldSerializationInfo):
        print(info)
        return diff.total_seconds()

    @field_serializer('diff2', mode='wrap')
    @staticmethod
    def diff2_ser(diff2: timedelta, nxt: SerializerFunctionWrapHandler, info: FieldSerializationInfo):
        value = nxt(diff2)
        return f"{value}postprocess"

m = WithCustomEncoders(
    dt=datetime(2032, 6, 1, tzinfo=timezone.utc), 
    diff=timedelta(minutes=2),
    diff2=timedelta(minutes=1)
)
print(m.model_dump_json())
# {"dt":1969660800.0,"diff":120.0,"diff2":"PT60Spostprocess"}

第2步：使用 @model_serializer 重构整体模型结构

目的：拦截整个模型的序列化过程，返回全新的字典或结构。
入口：定义在 BaseModel 内部。
动作：使用 @model_serializer(mode='plain|wrap') 装饰方法。
重要参数：mode='wrap' 时，需通过 nxt: SerializerFunctionWrapHandler 获取 Pydantic 默认生成的字典，再进行合并或覆盖。
完成状态：返回的字典直接替代默认序列化结果。

from typing import Dict, Any
from pydantic import BaseModel, model_serializer
from pydantic_core.core_schema import SerializerFunctionWrapHandler, SerializationInfo

class Model(BaseModel):
    x: str

    @model_serializer
    def ser_model(self, info: SerializationInfo):
        print(info)
        return {'x': f'xxxxxx {self.x}'}

    @model_serializer(mode='wrap')
    def ser_model_wrap(self, nxt: SerializerFunctionWrapHandler, info: SerializationInfo) -> Dict[str, Any]:
        print(info)
        return {'x': f'serialized {nxt(self)}'}

print(Model(x='test value').model_dump_json())
# {"x":"serialized {'x': 'test value'}"}

基于类型的统一序列化配置

在 Pydantic V1 版本中，开发者习惯在 ConfigDict 中配置 json_encoders 映射表。V2 版本已弃用该方式，推荐采用类型级别（Type-level）的注解方案。通过 typing.Annotated 结合 WrapSerializer 或 PlainSerializer，可为指定类型全局绑定序列化策略，避免重复编写装饰器。

以下示例展示了如何为整数类型分别配置格式化逗号分隔和乘法转换，并通过 when_used='json' 限定仅在导出 JSON 时生效：

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

def ser_wrap(v: Any, nxt: SerializerFunctionWrapHandler) -> str:
    return f'{nxt(v + 1):,}'

FancyInt = Annotated[int, WrapSerializer(ser_wrap, when_used='json')]
DoubleInt = Annotated[int, PlainSerializer(lambda x: x * 2)]

class MyModel(BaseModel):
    x: FancyInt
    y: DoubleInt

print(MyModel(x=1234, y=2).model_dump())
# {'x': 1234, 'y': 4}
print(MyModel(x=1234, y=2).model_dump(mode='json'))
# {'x': '1,235', 'y': 4}

高级特性：声明类型还原与精确过滤

序列化过程中常面临类型多态与字段隐私控制问题，Pydantic 提供了针对性工具。

第3步：切换实际类型序列化（SerializeAsAny）

目的：解决继承或多态场景中，Pydantic 默认按声明基类类型序列化导致子模型特有字段丢失的问题。
入口：导入 SerializeAsAny 包装类型声明。
动作：将属性类型从 BaseModel 改为 SerializeAsAny[BaseModel]。
重要参数：无额外参数，依赖运行时类型推断。
完成状态：序列化输出包含派生类的完整字段，而非仅保留父类定义字段。

from pydantic import BaseModel, SerializeAsAny

class User(BaseModel):
    name: str

class UserLogin(User):
    password: str

class OuterModel(BaseModel):
    # 声明为User类型，按照User类序列化，只有name字段
    user: User
    # 声明为Any包装，序列化时保留实际类型的所有字段
    user1: SerializeAsAny[User] = UserLogin(name='serialize as any', password='hunter')

# 实际类型为UserLogin
user = UserLogin(name='pydantic', password='hunter2')
m = OuterModel(user=user)
print(m)
# user=UserLogin(name='pydantic', password='hunter2') user1=UserLogin(name='serialize as any', password='hunter')
print(m.model_dump())
# {'user': {'name': 'pydantic'}, 'user1': {'name': 'serialize as any', 'password': 'hunter'}}

第4步：配置 exclude 与 include 精确控制字段

目的：按需剥离或保留部分字段，常用于日志脱敏或接口精简。
入口：model_dump(exclude={...}) 或 model_dump(include={...})。
动作：支持集合（Set）快速排除整块属性，支持字典（Dict）进行深度嵌套控制，甚至可通过索引定位列表元素（如 {'items': {0: True}}）。
重要参数：True 表示应用规则，'__all__' 表示通配子节点。
完成状态：返回的字典严格遵循过滤条件，不包含被拦截的结构。

from pydantic import BaseModel, SecretStr

class User(BaseModel):
    id: int
    username: str
    password: SecretStr

class Transaction(BaseModel):
    id: str
    user: User
    value: int

t = Transaction(
    id='1234567890',
    user=User(id=42, username='JohnDoe', password='hashedpassword'),
    value=9876543210,
)
# 使用集合排除整节点
print(t.model_dump(exclude={'user', 'value'}))
# {'id': '1234567890'}
# 使用字典深度排除
print(t.model_dump(exclude={'user': {'username', 'password'}, 'value': True}))
# {'id': '1234567890', 'user': {'id': 42}}
print(t.model_dump(include={'id': True, 'user': {'id'}}))
# {'id': '1234567890', 'user': {'id': 42}}

pickle 二进制序列化注意事项

除了 JSON 或字典，Pydantic 模型默认支持标准库 pickle 进行内存持久化。对于结构简单、不依赖外部状态或复杂回调的模型，可直接调用 pickle.dumps() 和 loads() 进行字节流转换。若模型中包含动态连接池或自定义不可序列化资源，反序列化可能失败，此时建议优先使用标准的 JSON 序列化配合重建逻辑。

# TODO need to get pickling to work
import pickle
from pydantic import BaseModel

class FooBarModel(BaseModel):
    a: str
    b: int

m = FooBarModel(a='hello', b=123)
print(m)
# > a='hello' b=123
data = pickle.dumps(m)
print(data[:20])
# > b'\x80\x04\x95\x95\x00\x00\x00\x00\x00\x00\x00\x8c\x08__main_'
m2 = pickle.loads(data)
print(m2)
# > a='hello' b=123

总结

以上就是 Python 中 Pydantic 序列化机制与高级配置的详细内容，涵盖了从基础转换到字段定制、类型还原及精准过滤的核心操作。关于 Pydantic 模型验证与异步数据处理的更多资料请关注本站其它相关文章！