Multiprocessing and event, type hint issue python

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

Multiprocessing and event, type hint issue python

问题

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

  1. # module_a.py
  2. from multiprocessing import Event
  3. from multiprocessing import Process
  4. class BackgroundWorker(Process):
  5. """创建一个后台工作进程。"""
  6. def __init__(
  7. self,
  8. name: str,
  9. daemon: bool,
  10. contents: Any,
  11. event: Event,
  12. ) -> None:
  13. """初始化默认值。"""
  14. self.contents: Any = contents
  15. self._event: Event = event
  16. super().__init__(name=name, daemon=daemon)
  17. def run(self) -> None:
  18. """运行目标函数。"""
  19. 一些代码
  20. if self._event.wait(0.4):
  21. 一些代码
  22. if self._event.is_set():
  23. break
  24. 一些代码
  1. # module_b.py
  2. from multiprocessing import Event
  3. event: Event = Event()
  4. def cancel_task():
  5. event.set()

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

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

请提出建议。

英文:

I have used multiprocessing module to perform a background task.

  1. # module_a.py
  2. from multiprocessing import Event
  3. from multiprocessing import Process
  4. class BackgroundWorker(Process):
  5. """Create a worker background process."""
  6. def __init__(
  7. self,
  8. name: str,
  9. daemon: bool,
  10. contents: Any,
  11. event: Event,
  12. ) -> None:
  13. """Initialize the defaults."""
  14. self.contents: Any = contents
  15. self._event: Event = event
  16. super().__init__(name=name, daemon=daemon)
  17. def run(self) -> None:
  18. """Run the target function."""
  19. some code
  20. if self._event.wait(0.4):
  21. some code
  22. if self._event.is_set():
  23. break
  24. some code
  1. # module_b.py
  2. from multiprocessing import Event
  3. event: Event = Event()
  4. def cancel_task():
  5. event.set()

Following are the errors on running mypy

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

Please suggest.

答案1

得分: 1

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

  1. # module_a.py
  2. from multiprocessing.synchronize import Event as EventClass
  3. Event 替换为 EventClass
  4. 例如
  5. self._event: EventClass = event
  1. # module_b.py
  2. import multiprocessing
  3. from multiprocessing.synchronize import Event as EventClass
  4. event: EventClass = multiprocessing.Event()
  5. def cancel_task():
  6. event.set()
英文:

As suggested by @Aaron, the new changes as below

  1. # module_a.py
  2. from multiprocessing.synchronize import Event as EventClass
  3. replace Event with EventClass
  4. for eg,
  5. self._event: EventClass = event
  1. # module_b.py
  2. import multiprocessing
  3. from multiprocessing.synchronize import Event as EventClass
  4. event: EventClass = multiprocessing.Event()
  5. def cancel_task():
  6. 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:

确定