Python多进程无法pickle本地对象问题求助

问题场景 #

我在构建测试运行器时，通过元类给测试方法自动添加before和after执行逻辑，使用multiprocessing调用这些修改后的方法时，触发了pickle序列化错误：

AttributeError: Can't pickle local object 'MetaTestCase.__new__.<locals>.replaced_func.<locals>.new_test'

尝试用pathos.helpers.mp.Process绕过pickle限制后，又出现索引错误：

Process Process-1:
Traceback (most recent call last):
    result = fn(*args, **kwargs)
IndexError: tuple index out of range

相关代码如下：

父元类 #

from typing import Tuple

class MetaTestCase(type):
    def __new__(cls, name: str, bases: Tuple, attrs: dict):
        def replaced_func(fn):
            def new_test(*args, **kwargs):
                args[0].before()
                result = fn(*args, **kwargs)
                args[0].after()
                return result
            return new_test

        # 替换以test开头的方法
        for i in attrs:
            if callable(attrs[i]) and attrs[i].__name__.startswith('test'):
                attrs[i] = replaced_func(attrs[i])

        return super(MetaTestCase, cls).__new__(cls, name, bases, attrs)

继承元类的子类 #

class TestCase(metaclass=MetaTestCase):
    def before(self) -> None:
        """可重写，测试前执行"""
        pass

    def after(self) -> None:
        """可重写，测试后执行"""
        pass

测试类 #

class TestApi(TestCase):
    def before(self):
        print('before')

    def after(self):
        print('after')

    def test_api_one(self):
        print('test')

原调用代码 #

import importlib
from multiprocessing import Process

module = importlib.import_module('tests.api.test_api')  # 导入模块
TestApi = getattr(module, 'TestApi')  # 获取类

process = Process(target=getattr(TestApi, 'test_api_one')) # 调用类方法
process.start()
process.join()

解决方案 #

方案1：将包装函数移出元类作用域 #

pickle无法序列化局部函数（嵌套在__new__里的new_test），把包装函数移到顶层全局作用域即可解决：

from typing import Tuple

def replaced_func(fn):
    def new_test(*args, **kwargs):
        args[0].before()
        result = fn(*args, **kwargs)
        args[0].after()
        return result
    return new_test

class MetaTestCase(type):
    def __new__(cls, name: str, bases: Tuple, attrs: dict):
        # 替换以test开头的方法
        for attr_name in attrs:
            if callable(attrs[attr_name]) and attr_name.startswith('test'):
                attrs[attr_name] = replaced_func(attrs[attr_name])
        return super().__new__(cls, name, bases, attrs)

方案2：修改多进程调用逻辑，传递实例方法 #

原调用直接传递类的未绑定方法，导致方法缺少self参数，引发索引错误。先实例化测试类，再传递实例的绑定方法：

import importlib
from multiprocessing import Process

module = importlib.import_module('tests.api.test_api')
TestApi = getattr(module, 'TestApi')
test_instance = TestApi()  # 实例化测试类

process = Process(target=test_instance.test_api_one)
process.start()
process.join()

方案3：用类装饰器替代元类（可选） #

如果不想依赖元类，可改用类装饰器实现相同的before/after逻辑，避免元类带来的序列化问题：

def test_case_decorator(cls):
    for attr_name in dir(cls):
        attr = getattr(cls, attr_name)
        if callable(attr) and attr_name.startswith('test'):
            def wrapper(fn):
                def new_test(self, *args, **kwargs):
                    self.before()
                    result = fn(self, *args, **kwargs)
                    self.after()
                    return result
                return new_test
            setattr(cls, attr_name, wrapper(attr))
    return cls

@test_case_decorator
class TestCase:
    def before(self) -> None:
        pass
    def after(self) -> None:
        pass

错误原因说明 #

1. pickle序列化错误：new_test是MetaTestCase.__new__内部的嵌套函数，属于局部对象，pickle无法序列化没有全局可引用名称的对象。
2. 索引错误：直接传递类的未绑定方法时，多进程调用不会自动传入实例（self），导致args[0]不存在，触发索引越界。

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