使用Pydantic对象

使用Pydantic的dataclass

使用TypeAdapter与从数据生成dataclass类

from typing import Generic, TypeVar, List
from pydantic.dataclasses import dataclass
from pydantic import TypeAdapter

T = TypeVar("T")

@dataclass
class Item:
    id: int
    name: str

@dataclass
class Response(Generic[T]):
    status: str
    data: List[T]  # 嵌套泛型列表

# 模拟嵌套字典数据
raw_data = {
    "status": "success",
    "data": [{"id": 1, "name": "Apple"}, {"id": 2, "name": "Banana"}]
}

# 实例化（Pydantic 会自动转换嵌套内容）

adapter = TypeAdapter(Response[Item])
res = adapter.validate_python(raw_data)

print(res.data[0].name)  # 输出: Apple
# 此时输入 res. ，你只会看到 status 和 data，没有 BaseModel 的那堆方法

使用使用TypeAdapter与从dataclass类生成数据

from pydantic.dataclasses import dataclass
from pydantic import TypeAdapter

@dataclass
class User:
    id: int
    name: str

user = User(id=1, name="Alice")

# 使用 TypeAdapter 进行转化
adapter = TypeAdapter(User)
user_dict = adapter.dump_python(user)

print(user_dict)  # {'id': 1, 'name': 'Alice'}

使用__post_init__方法在对象初始化完成后自动计算其他数值

@dataclass
class XCopyTask:
    no: int = Field(alias="序号")
    src_address: str = Field(alias="源库连接地址")
    src_schema: str = Field(alias="源库schema名称")
    src_table: str = Field(alias="源表名称")
    tgt_address: str = Field(alias="目标库连接地址")
    tgt_schema: str = Field(alias="目标库schema名称")
    tgt_table: str = Field(alias="目标库表名称")
    data_size: str = Field(alias="源库表数据量级")
    sync_type: str = Field(alias="同步方式")
    where: str | None = Field(alias="增量同步条件")
    freq: str = Field(alias="同步频率-时间")
    influence: str = Field(alias="是否影响出单业务")
    impact: str = Field(alias="是否包含客户敏感信息")
    src_host: str = ""
    src_port: str = ""
    src_dbname: str = ""
    tgt_host: str = ""
    tgt_port: str = ""
    tgt_dbname: str = ""

    def __post_init__(self):
        src_search = address_pattern.search(self.src_address)
        if src_search:
            self.src_host, self.src_port, self.src_dbname = src_search.groups()

        tgt_search = address_pattern.search(self.tgt_address)
        if tgt_search:
            self.tgt_host, self.tgt_port, self.tgt_dbname = tgt_search.groups()

从内容生成模型文件

使用官方库datamodel-code-generator 来通过数据文件生成对应的python模型。

生成命令：datamodel-codegen

也可以在pyproject.toml文件中直接编辑参数

[tool.datamodel-codegen]
input = "data.json" # 数据文件
output = "models.py" # 模型文件
output-model-type = "pydantic_v2.BaseModel" # 继承的类
input-file-type = "json" # 输入文件类型
target-python-version = "3.12" # 输出模型的python版本