使用 TypeVar 作为 Callable 的参数时出现问题

Question

我遇到了以下代码的问题，至少根据 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]}")

当我使用 mypy 检查它时，出现以下错误：

error: Argument 1 to "imap_unordered" of "Pool" has incompatible type "Callable[[T], Tuple[T, bool]]"; expected "Callable[[int], Tuple[T, bool]]"

实际类型显然应该满足预期类型。 我做错了什么或者是 mypy 或 multiprocessing.Pool 中的错误吗？

提前感谢您的任何评论。

Answer 1

如 Jasmijn 的评论所示，这是一个最小的示例：

from typing import TypeVar

T = TypeVar("T")

def do_something(a: T) -> T:
    return a

map(do_something, [1])

此操作失败并出现相同的错误。我认为这是

mypy

的限制，但我们可以通过引入额外的辅助函数来修复它，通过该函数我们显式提供类型，而不是

mypy

必须推断它。

def new_do_something(t: Type[T]) -> Callable[[T], T]:
    return do_something

使用上述内容，这不再导致任何错误，同时仍然允许

do_something

是通用的：

map(new_do_something(int), [1])

对于您的原始示例，修改后的版本将如下所示：

from multiprocessing import Pool
from typing import Tuple, TypeVar, Type, Callable

T = TypeVar("T", int, str)


def do_something(a: T) -> Tuple[T, bool]:
    # Something happening here...
    return a, False

def new_do_something(t: Type[T]) -> Callable[[T], Tuple[T, bool]]:
    return do_something

if __name__ == "__main__":
    pool = Pool(processes=10)
    nums = list(range(100000, 1000000))
    for r in pool.imap_unordered(new_do_something(int), nums):
        if r[1]:
            print(f"Got {r[0]}")

Answer 2

这是一个替代解决方案...

我们将

do_something

转换为具有可调用实例的泛型类：

from multiprocessing import Pool
from typing import Generic, Tuple, TypeVar

T = TypeVar("T", int, str)


class do_something(Generic[T]):
    def __call__(self, 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[int](), nums):
        if r[1]:
            print(f"Got {r[0]}")

看起来并没有好多少，但至少避免了定义额外的中间函数。