You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

求支持抽象基类及用户自定义子类的Python对象序列化库

适配抽象基类与用户自定义子类的Python序列化方案

1. attrs + cattrs

attrs可替代或兼容dataclass,cattrs作为配套序列化工具,天生支持多态场景,对用户自定义子类友好:

  • 用户自定义子类后,仅需调用cattrs.register_structure_hook和cattrs.register_unstructure_hook即可让库识别子类
  • 完全不干涉对象原生实例化逻辑,兼容dataclass/attrs类的正常使用
  • 示例代码:
from attrs import define
from cattrs import structure, unstructure, register_structure_hook, register_unstructure_hook
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List

# 抽象基类
class ComponentType(ABC):
    @abstractmethod
    def do_something(self):
        pass

# 用户自定义子类
@define
class CustomComponent(ComponentType):
    value: int
    def do_something(self):
        print(self.value)

# 顶层dataclass
@dataclass
class ExampleProgram:
    component_1: ComponentType
    widgets: Dict[str, WidgetType]  # WidgetType为抽象基类
    foo: List[FooType]              # FooType为抽象基类

# 注册序列化钩子:嵌入类型标识
def unstructure_component(obj: ComponentType):
    data = unstructure(obj)
    data["_type"] = f"{obj.__module__}.{obj.__class__.__name__}"
    return data

# 注册反序列化钩子:根据类型标识匹配子类
def structure_component(data: dict, cls: type[ComponentType]):
    type_path = data.pop("_type")
    module_name, cls_name = type_path.rsplit(".", 1)
    import importlib
    module = importlib.import_module(module_name)
    subclass = getattr(module, cls_name)
    return structure(data, subclass)

register_structure_hook(ComponentType, structure_component)
register_unstructure_hook(ComponentType, unstructure_component)
# 同理为WidgetType、FooType注册钩子

# 使用示例
custom_comp = CustomComponent(42)
prog = ExampleProgram(component_1=custom_comp, widgets={}, foo=[])

serialized = unstructure(prog)
deserialized = structure(serialized, ExampleProgram)

2. marshmallow + marshmallow-polymorphic

marshmallow是老牌序列化库,marshmallow-polymorphic扩展专门处理多态序列化:

  • 抽象基类的Schema配置类型鉴别字段,用户子类只需继承对应Schema并注册到父类Schema
  • Schema与业务类完全分离,不影响原类实例化
  • 示例代码:
from marshmallow import Schema, fields
from marshmallow_polymorphic import PolymorphicSchema, PolymorphicField
from abc import ABC, abstractmethod
from dataclasses import dataclass, asdict
from typing import Dict, List

# 抽象基类
class ComponentType(ABC):
    @abstractmethod
    def do_something(self):
        pass

# 用户自定义子类
@dataclass
class CustomComponent(ComponentType):
    value: int
    def do_something(self):
        print(self.value)

# 定义多态Schema
class ComponentTypeSchema(PolymorphicSchema):
    type_field = "_type"  # 用于识别子类的字段
    type_schemas = {}     # 子类Schema注册表

class CustomComponentSchema(Schema):
    value = fields.Int()

# 注册子类Schema到父类
ComponentTypeSchema.type_schemas["CustomComponent"] = CustomComponentSchema()

# 顶层类Schema
class ExampleProgramSchema(Schema):
    component_1 = PolymorphicField(ComponentTypeSchema)
    widgets = fields.Dict(values=PolymorphicField(WidgetTypeSchema))  # 适配WidgetType
    foo = fields.List(PolymorphicField(FooTypeSchema))                # 适配FooType

# 使用示例
custom_comp = CustomComponent(42)
prog = ExampleProgram(component_1=custom_comp, widgets={}, foo=[])

schema = ExampleProgramSchema()
serialized = schema.dump(asdict(prog))
deserialized_data = schema.load(serialized)
deserialized_prog = ExampleProgram(**deserialized_data)

3. 基于pickle的自定义逻辑(无第三方库)

若不想引入外部依赖,可基于pickle扩展(仅在可信环境使用,注意安全风险):

  • 在抽象基类中实现__getstate__和__setstate__方法,手动嵌入类型标识
  • 反序列化时根据标识动态加载子类
  • 示例代码:
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Dict, List
import pickle

class ComponentType(ABC):
    @abstractmethod
    def do_something(self):
        pass

    def __getstate__(self):
        state = self.__dict__.copy()
        state["_type"] = f"{self.__module__}.{self.__class__.__name__}"
        return state

    def __setstate__(self, state):
        type_path = state.pop("_type")
        module_name, cls_name = type_path.rsplit(".", 1)
        import importlib
        module = importlib.import_module(module_name)
        self.__class__ = getattr(module, cls_name)
        self.__dict__.update(state)

@dataclass
class CustomComponent(ComponentType):
    value: int
    def do_something(self):
        print(self.value)

@dataclass
class ExampleProgram:
    component_1: ComponentType
    widgets: Dict[str, WidgetType]
    foo: List[FooType]

# 使用示例
custom_comp = CustomComponent(42)
prog = ExampleProgram(component_1=custom_comp, widgets={}, foo=[])

serialized = pickle.dumps(prog)
deserialized = pickle.loads(serialized)

关键注意点

  • 所有方案的核心逻辑一致:序列化时嵌入子类的完整类型路径,反序列化时根据路径加载对应子类
  • 需为用户提供简单的子类注册方式(如调用钩子函数、注册Schema),确保序列化库能识别自定义子类
  • 优先推荐attrs + cattrs组合,对dataclass的支持最原生,代码侵入性最低

内容的提问来源于stack exchange,提问作者PrehensileOwl

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.27 22:47:55