Python类含动态Enum属性致Pathos ProcessingPool多进程序列化失败问题

问题复现 #

运行以下代码时会抛出PicklingError：

from enum import Enum
from pathos.multiprocessing import ProcessingPool

class MyClass:
    def __init__(self, group_dict):
        self.group_dict = group_dict
        self.tags_emum = Enum(
            value="MyEnum",
            names={v.upper(): v for v in self.group_dict.keys()},
            type=str,
        )

    def fnc1(self, names_list):
        pool = ProcessingPool(nodes=2)
        result = pool.map(self.fnc2, names_list)
        return result

    def fnc2(self, name):
        return len(name)

if __name__ == "__main__":
    inst = MyClass(group_dict={"key1": "val1", "key2": "val2"})
    print(inst.fnc1(names_list=["StackOverflow", "Python", "Question"]))

错误信息：

_pickle.PicklingError: Can't pickle <enum 'MyEnum'>: it's not found as __main__.MyEnum

移除self.tags_emum属性后，代码正常输出[13, 6, 8]。

1. 失败原因 #

Pickle序列化对象时，依赖类的模块级引用路径完成反序列化。这里的MyEnum是在MyClass.__init__方法内动态创建的，属于实例级的动态类，并没有注册在__main__模块的全局命名空间中。

当pathos的ProcessingPool将MyClass实例传递给子进程时，pickle会记录MyEnum的引用为__main__.MyEnum，但子进程启动后，其__main__模块中没有这个类的定义，因此无法完成反序列化，抛出PicklingError。

即使pathos底层使用dill（比标准pickle支持更多类型），动态创建的Enum类因为缺少模块级注册信息，依然无法被正确序列化。

2. 保留Enum属性的修复方案 #

方案一：将动态Enum注册到模块命名空间 #

在创建完动态Enum后，手动将其添加到__main__模块的全局变量中，让子进程能找到该类的引用：

from enum import Enum
import __main__
from pathos.multiprocessing import ProcessingPool

class MyClass:
    def __init__(self, group_dict):
        self.group_dict = group_dict
        self.tags_emum = Enum(
            value="MyEnum",
            names={v.upper(): v for v in self.group_dict.keys()},
            type=str,
        )
        # 将动态Enum注册到__main__模块
        __main__.MyEnum = self.tags_emum

    def fnc1(self, names_list):
        pool = ProcessingPool(nodes=2)
        result = pool.map(self.fnc2, names_list)
        return result

    def fnc2(self, name):
        return len(name)

if __name__ == "__main__":
    inst = MyClass(group_dict={"key1": "val1", "key2": "val2"})
    print(inst.fnc1(names_list=["StackOverflow", "Python", "Question"]))

方案二：自定义序列化/反序列化逻辑 #

通过实现__getstate__和__setstate__方法，序列化时只保存Enum的原始构建数据，反序列化时重新创建Enum，避免直接序列化动态类：

from enum import Enum
from pathos.multiprocessing import ProcessingPool

class MyClass:
    def __init__(self, group_dict):
        self.group_dict = group_dict
        self._init_tags_enum()

    def _init_tags_enum(self):
        # 抽离Enum初始化逻辑，方便复用
        self.tags_emum = Enum(
            value="MyEnum",
            names={v.upper(): v for v in self.group_dict.keys()},
            type=str,
        )

    def __getstate__(self):
        # 序列化时只保存必要数据，不保存Enum类本身
        state = self.__dict__.copy()
        del state['tags_emum']
        return state

    def __setstate__(self, state):
        # 反序列化时恢复数据并重新创建Enum
        self.__dict__.update(state)
        self._init_tags_enum()

    def fnc1(self, names_list):
        pool = ProcessingPool(nodes=2)
        result = pool.map(self.fnc2, names_list)
        return result

    def fnc2(self, name):
        return len(name)

if __name__ == "__main__":
    inst = MyClass(group_dict={"key1": "val1", "key2": "val2"})
    print(inst.fnc1(names_list=["StackOverflow", "Python", "Question"]))

方案三：改用模块级动态Enum（提前创建） #

如果group_dict可以提前确定，可在模块级别动态创建Enum，而不是在实例的__init__中：

from enum import Enum
from pathos.multiprocessing import ProcessingPool

# 提前定义group_dict（如果允许）
GROUP_DICT = {"key1": "val1", "key2": "val2"}
MyEnum = Enum(
    value="MyEnum",
    names={v.upper(): v for v in GROUP_DICT.keys()},
    type=str,
)

class MyClass:
    def __init__(self, group_dict=GROUP_DICT):
        self.group_dict = group_dict
        self.tags_emum = MyEnum

    def fnc1(self, names_list):
        pool = ProcessingPool(nodes=2)
        result = pool.map(self.fnc2, names_list)
        return result

    def fnc2(self, name):
        return len(name)

if __name__ == "__main__":
    inst = MyClass()
    print(inst.fnc1(names_list=["StackOverflow", "Python", "Question"]))

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