如何使用Pydantic 2将数据转换为指定JSON模型导出至Kafka子队列？

完全可以用Pydantic 2的原生功能实现这个需求，通过定义目标输出模型、利用计算字段和序列化器，就能安全高效地完成格式转换。

实现步骤及代码 #

1. 导入依赖

from pydantic import BaseModel, computed_field, field_serializer
from datetime import datetime

2. 定义原数据模型

（规范类名，Pydantic推荐类名首字母大写）

class Obj(BaseModel):
    resource: str
    resourceid: int
    resourceTime: datetime
    name: str
    option1: str
    option2: int

3. 定义目标输出模型

先定义attributes数组中的子模型，再定义最终导出用的模型：

class Attribute(BaseModel):
    name: str
    value: str | int  # 兼容字符串和整数类型的属性值

class KafkaExportModel(BaseModel):
    # 直接映射原模型的固定字段
    resource: str
    name: str
    resourceid: str
    resourceTime: datetime

    # 用计算字段动态生成attributes列表
    @computed_field(return_type=list[Attribute])
    def attributes(self) -> list[Attribute]:
        # 指定需要转换为attribute的字段名
        target_fields = ["option1", "option2", "name"]
        return [Attribute(name=field, value=getattr(self, field)) for field in target_fields]
    
    # 序列化resourceTime为指定格式的字符串
    @field_serializer('resourceTime')
    def format_resource_time(self, value: datetime) -> str:
        return value.strftime("%Y-%m-%d %H:%M:%S")
    
    # 把resourceid从整数转成字符串输出
    @field_serializer('resourceid')
    def convert_resource_id(self, value: int) -> str:
        return str(value)

4. 执行转换并导出

# 输入的原始JSON数据
input_json = {
    "resource": "TEST",
    "resourceid": "1234",
    "resourceTime": "2024-09-03 00:00:00",
    "option1": "test",
    "option2": 23421,
    "name": "test"
}

# 解析原始数据到原模型（Pydantic自动处理resourceid的字符串转整数）
original_obj = Obj(**input_json)

# 转换为目标导出模型
kafka_obj = KafkaExportModel(**original_obj.model_dump())

# 导出为符合要求的JSON字符串，直接发送到Kafka即可
output_json = kafka_obj.model_dump_json(indent=2)
print(output_json)

关键特性说明 #

• 计算字段（computed_field）：自动根据原模型字段生成attributes数组，无需手动拼接，保证类型安全。
• 字段序列化器（field_serializer）：精准控制字段的输出格式，比如日期格式化、数值类型转换。
• 类型自动转换：Pydantic 2会自动处理输入JSON中resourceid的字符串转整数，无需额外代码。

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