如何为第三方类添加Pydantic序列化/反序列化支持？

当你需要在Pydantic的BaseModel中使用不支持序列化的第三方类（且无法修改其源码）时，可以通过自定义序列化/反序列化函数，并结合Pydantic的字段处理器或自定义类型来实现自动转换。以下是具体实现步骤：

先定义基础的序列化/反序列化函数 #

假设我们已经针对Numpy数组实现了转换逻辑：

import numpy as np

# 将np.ndarray转为JSON可序列化的字典结构
def serialize_numpy_array(arr: np.ndarray) -> dict:
    return {
        'dtype': str(arr.dtype),
        'shape': arr.shape,
        'data': arr.tobytes().hex()
    }

# 从字典结构还原np.ndarray实例
def deserialize_numpy_array(data: dict) -> np.ndarray:
    dtype = np.dtype(data['dtype'])
    shape = data['shape']
    data_bytes = bytes.fromhex(data['data'])
    return np.frombuffer(data_bytes, dtype=dtype).reshape(shape)

方式一：字段级处理器（适合单个Model字段使用） #

直接在BaseModel中为目标字段添加序列化和验证器，指定调用自定义函数：

from pydantic import BaseModel, field_serializer, field_validator

class MySerializableClass(BaseModel):
    foreign_class_instance: np.ndarray

    # 序列化时自动调用自定义函数
    @field_serializer('foreign_class_instance')
    def serialize_foreign(self, value: np.ndarray) -> dict:
        return serialize_numpy_array(value)

    # 反序列化时先处理输入，再验证类型
    @field_validator('foreign_class_instance', mode='before')
    def deserialize_foreign(cls, value: dict | np.ndarray) -> np.ndarray:
        if isinstance(value, np.ndarray):
            return value
        return deserialize_numpy_array(value)

测试代码 #

# 序列化测试
test_arr = np.array([[1,2,3], [4,5,6]], dtype=np.int32)
model = MySerializableClass(foreign_class_instance=test_arr)
serialized_data = model.model_dump()
# 输出：{'foreign_class_instance': {'dtype': 'int32', 'shape': (2, 3), 'data': '010000000200000003000000040000000500000006000000'}}

# 反序列化测试
restored_model = MySerializableClass.model_validate(serialized_data)
print(restored_model.foreign_class_instance)
# 输出：[[1 2 3]
#        [4 5 6]]

方式二：自定义Pydantic类型（适合多处重复使用） #

如果该第三方类需要在多个BaseModel中使用，自定义专属类型可以避免重复代码：

from pydantic import GetCoreSchemaHandler
from pydantic_core import core_schema
from typing import Any

class NumpyArrayType(np.ndarray):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        # 定义序列化规则
        serialize_schema = core_schema.plain_serializer_function_ser_schema(
            serialize_numpy_array,
            return_schema=core_schema.dict_schema()
        )
        # 定义反序列化规则
        deserialize_schema = core_schema.no_info_plain_validator_function(deserialize_numpy_array)
        
        # 组合完整Schema：支持直接传入数组或序列化后的字典，输出数组
        return core_schema.json_or_python_schema(
            python_schema=core_schema.is_instance(np.ndarray),
            json_schema=deserialize_schema,
            serialization=serialize_schema,
        )

# 在Model中直接使用自定义类型
class MySerializableClass(BaseModel):
    foreign_class_instance: NumpyArrayType

核心说明 #

• mode='before'确保反序列化时先将JSON字典转换为第三方类实例，再进行Pydantic的类型检查。
• 对于其他第三方类，只需替换示例中的serialize_numpy_array和deserialize_numpy_array为对应的自定义转换函数即可，逻辑完全通用。
• 上述示例基于Pydantic v2，若使用v1版本，可替换为@validator装饰器和自定义json_encoder。

内容的提问来源于stack exchange，提问作者bluenote10