Python Multiprocessing Event类型提示报错问题求助

问题场景 #

我用multiprocessing模块实现后台任务，代码如下：

module_a.py #

from multiprocessing import Event
from multiprocessing import Process
from typing import Any

class BackgroundWorker(Process):
    """Create a worker background process."""

    def __init__(
        self,
        name: str,
        daemon: bool,
        contents: Any,
        event: Event,
    ) -> None:
        """Initialize the defaults."""
        self.contents: Any = contents
        self._event: Event = event
        super().__init__(name=name, daemon=daemon)

    def run(self) -> None:
        """Run the target function."""
        # some code
        
        if self._event.wait(0.4):
            # some code
            pass
        
        if self._event.is_set():
            break
            
        # some code

module_b.py #

from multiprocessing import Event

event: Event = Event()

def cancel_task():
    event.set()

运行mypy时触发以下错误：

error: Variable "multiprocessing.Event" is not valid as a type [valid-type]
note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
error: Variable "multiprocessing.Event" is not valid as a type [valid-type]
note: See https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
error: Event? has no attribute "is_set" [attr-defined]
error: Event? has no attribute "wait" [attr-defined]

问题根源 #

直接导入的multiprocessing.Event是工厂函数，用来创建事件实例，但它本身不是一个类型。实际生成的事件对象属于multiprocessing.synchronize.Event类，Mypy无法将工厂函数识别为有效的类型注解，所以抛出错误。

修复方案 #

方案1：导入正确的类型类 #

直接从multiprocessing.synchronize导入Event类型，替换原有的导入语句：

修改后的module_a.py

from multiprocessing import Process
from multiprocessing.synchronize import Event
from typing import Any

class BackgroundWorker(Process):
    """Create a worker background process."""

    def __init__(
        self,
        name: str,
        daemon: bool,
        contents: Any,
        event: Event,
    ) -> None:
        """Initialize the defaults."""
        self.contents: Any = contents
        self._event: Event = event
        super().__init__(name=name, daemon=daemon)

    def run(self) -> None:
        """Run the target function."""
        # some code
        
        if self._event.wait(0.4):
            # some code
            pass
        
        if self._event.is_set():
            break
            
        # some code

修改后的module_b.py

# 用别名区分工厂函数和类型
from multiprocessing import Event as EventFactory
from multiprocessing.synchronize import Event

event: Event = EventFactory()

def cancel_task():
    event.set()

方案2：使用类型别名（Python 3.10+） #

如果你的Python版本是3.10及以上，可以用TypeAlias明确标注类型：

from multiprocessing import Event
from typing import TypeAlias

# 定义类型别名
EventType: TypeAlias = Event

# 后续用EventType做注解
event: EventType = Event()

方案3：临时兼容方案（不推荐） #

如果暂时无法调整导入结构，也可以用type(Event())作为类型注解，但这种写法不够直观：

from multiprocessing import Event

event: type(Event()) = Event()

验证 #

修改完成后重新运行mypy，所有错误应该都会被解决。

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