我遇到了以下代码的问题,至少根据 mypy 是这样的:
from multiprocessing import Pool
from typing import Tuple, TypeVar
T = TypeVar("T", int, str)
def do_something(a: T) -> Tuple[T, bool]:
# Something happening here...
return a, False
if __name__ == "__main__":
pool = Pool(processes=10)
nums = list(range(100000, 1000000))
for r in pool.imap_unordered(do_something, nums):
if r[1]:
print(f"Got {r[0]}")
Run Code Online (Sandbox Code Playgroud)
当我使用 mypy 检查它时,出现以下错误:
error: Argument 1 to "imap_unordered" of "Pool" has incompatible type "Callable[[T], Tuple[T, bool]]"; expected "Callable[[int], Tuple[T, bool]]"
Run Code Online (Sandbox Code Playgroud)
实际类型显然应该满足预期类型。我做错了什么还是 mypy 或 multiprocessing.Pool 中的错误?
预先感谢您的任何评论。