Multiprocessing and event, type hint issue python

huangapple go评论68阅读模式
英文:

Multiprocessing and event, type hint issue python

问题

我使用了multiprocessing模块来执行后台任务。

# module_a.py

from multiprocessing import Event
from multiprocessing import Process

class BackgroundWorker(Process):
    """创建一个后台工作进程。"""

    def __init__(
        self,
        name: str,
        daemon: bool,
        contents: Any,
        event: Event,
    ) -> None:
        """初始化默认值。"""
        self.contents: Any = contents
        self._event: Event = event
        super().__init__(name=name, daemon=daemon)

    def run(self) -> None:
        """运行目标函数。"""
        一些代码
        
        if self._event.wait(0.4):
        
        一些代码
        
        if self._event.is_set():
            break
        	
        一些代码
# module_b.py

from multiprocessing import Event

event: Event = Event()

def cancel_task():
    event.set()     

在运行mypy时出现以下错误:

error: 变量 "multiprocessing.Event" 无效作为类型 [valid-type]
note: 请参考 https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
error: 变量 "multiprocessing.Event" 无效作为类型 [valid-type]
note: 请参考 https://mypy.readthedocs.io/en/stable/common_issues.html#variables-vs-type-aliases
error: Event? 没有属性 "is_set" [attr-defined]
error: Event? 没有属性 "wait" [attr-defined]

请提出建议。

英文:

I have used multiprocessing module to perform a background task.

# module_a.py

from multiprocessing import Event
from multiprocessing import Process

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
        
        if self._event.is_set():
        	break
        	
       some code
# module_b.py

from multiprocessing import Event

event: Event = Event()

def cancel_task():
    event.set()     

Following are the errors on running 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]

Please suggest.

答案1

得分: 1

根据 @Aaron 的建议,以下是新的更改:

# module_a.py

from multiprocessing.synchronize import Event as EventClass

将 Event 替换为 EventClass
例如
self._event: EventClass = event

# module_b.py

import multiprocessing
from multiprocessing.synchronize import Event as EventClass

event: EventClass = multiprocessing.Event()

def cancel_task():
    event.set()  
英文:

As suggested by @Aaron, the new changes as below

# module_a.py

from multiprocessing.synchronize import Event as EventClass

replace Event with EventClass
for eg, 
self._event: EventClass = event

# module_b.py

import multiprocessing
from multiprocessing.synchronize import Event as EventClass

event: EventClass = multiprocessing.Event()

def cancel_task():
    event.set()  

huangapple
  • 本文由 发表于 2023年3月4日 01:22:05
  • 转载请务必保留本文链接:https://go.coder-hub.com/75630114.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定