使用Python Multiprocessing处理Pydantic BaseModel报错求助

你遇到的AttributeError: '__signature__' attribute of 'data_object' is class-only错误，是因为Pydantic的BaseModel自带的__signature__是类专属属性，multiprocessing.Manager在生成代理对象时会尝试访问该属性，而代理机制无法正确处理这类类属性导致的。下面提供两种可行的解决思路：

方法一：用包装类封装Pydantic对象 #

通过自定义一个简单的包装类，把Pydantic Model实例包裹起来，让Manager注册这个包装类，避免直接操作Pydantic Model类。包装类提供 getter/setter 方法来操作内部的Pydantic对象：

import pydantic
from typing import Optional
import multiprocessing
from multiprocessing.managers import BaseManager

class DataObject(pydantic.BaseModel):
    url: str
    downloaded: Optional[bool] = False

# 包装类用于进程间共享
class SharedDataObject:
    def __init__(self, url: str, downloaded: Optional[bool] = False):
        self.model = DataObject(url=url, downloaded=downloaded)
    
    def get_url(self):
        return self.model.url
    
    def set_downloaded(self, value: bool):
        self.model.downloaded = value
    
    def get_downloaded(self):
        return self.model.downloaded

class CustomManager(BaseManager):
    pass

def downloader(single_data: SharedDataObject):
    single_data.set_downloaded(True)

if __name__ == '__main__':
    # 单进程测试（正常运行）
    just_one_object = DataObject(url='url1')
    print(just_one_object.downloaded)
    just_one_object.downloaded = True
    print(just_one_object.downloaded)

    # 多进程共享逻辑
    CustomManager.register('SharedDataObject', SharedDataObject)
    CustomManager.register('list', list)
    with CustomManager() as manager:
        shared_single_object = manager.SharedDataObject(url='url2')
        print(shared_single_object.get_downloaded())
        downloader(shared_single_object)
        print(shared_single_object.get_downloaded())

        managed_list = manager.list([manager.SharedDataObject(url=f'url{v}') for v in range(5)])

        pool = multiprocessing.Pool(processes=5)
        pool.map(downloader, managed_list)
        pool.close()
        pool.join()

        # 输出最终状态
        for item in managed_list:
            print(f"URL: {item.get_url()}, Downloaded: {item.get_downloaded()}")

方法二：避免共享对象，用返回结果更新状态 #

放弃进程间共享对象的思路，改用Pool.map的返回值来获取处理后的Pydantic对象。这种方式更简单，也避免了共享对象带来的同步问题：

import pydantic
from typing import Optional
import multiprocessing

class DataObject(pydantic.BaseModel):
    url: str
    downloaded: Optional[bool] = False

def downloader(single_data: DataObject) -> DataObject:
    # 模拟下载操作，修改状态
    single_data.downloaded = True
    return single_data

if __name__ == '__main__':
    # 单进程测试
    just_one_object = DataObject(url='url1')
    print(just_one_object.downloaded)
    just_one_object = downloader(just_one_object)
    print(just_one_object.downloaded)

    # 多进程批量处理
    task_list = [DataObject(url=f'url{v}') for v in range(5)]

    pool = multiprocessing.Pool(processes=5)
    result_list = pool.map(downloader, task_list)
    pool.close()
    pool.join()

    # 打印处理后的结果
    for item in result_list:
        print(f"URL: {item.url}, Downloaded: {item.downloaded}")

错误原因说明 #

Pydantic的BaseModel类会自动生成__signature__类属性，用于参数校验和类型提示。multiprocessing.Manager在注册类并生成跨进程代理对象时，会尝试复制类的属性，但__signature__是仅属于类的属性，无法被代理对象正确访问，因此触发了AttributeError。

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