如何为并行执行多函数的Python函数添加类型提示？

核心解决方案：使用可变参数泛型（Python 3.10+） #

要实现支持任意数量无参Callable、且完整保留返回值类型信息的parallelize函数，最简洁的方式是利用Python 3.10引入的可变参数泛型（TypeVarTuple和Unpack），无需手动编写大量@overload声明。

完整实现代码 #

from typing import Callable, TypeVarTuple, Unpack, Tuple
from concurrent.futures import ThreadPoolExecutor, wait, Future

# 定义可变类型元组，用于捕获任意数量的返回值类型
Ts = TypeVarTuple('Ts')

def parallelize(*callables: Unpack[Tuple[Callable[[], Unpack[Ts]]]]) -> Tuple[Unpack[Ts]]:
    with ThreadPoolExecutor() as executor:
        # 提交所有并行任务
        futures: list[Future[Unpack[Ts]]] = [executor.submit(call) for call in callables]
        # 等待所有任务执行完成
        wait(futures)
        # 按提交顺序收集结果，返回对应类型的元组
        return tuple(future.result() for future in futures)

关键细节说明 #

1. 类型标注逻辑：

   • TypeVarTuple('Ts')定义一个可变的类型集合，用来逐个捕获传入的每个Callable的返回值类型。
   • *callables: Unpack[Tuple[Callable[[], Unpack[Ts]]]]表示接受任意数量的无参Callable，每个Callable的返回值类型对应Ts中的一个元素。
   • 返回值Tuple[Unpack[Ts]]确保返回元组的类型与传入Callable的返回值类型严格一一对应，类型检查工具（如mypy）能准确推断每个变量的类型。

2. 并行实现优化：

   • 使用concurrent.futures.wait(futures)统一等待所有任务完成，比逐个调用result()更高效，也能统一管理任务状态。
   • 返回tuple(...)而非生成器，避免原代码中返回生成器导致的类型推断失效、迭代一次后无法复用的问题。

使用示例 #

def get_airflow() -> str:
    return "a"
def get_prometheus() -> dict[str, str]:
    return {"b": "c"}
def get_zabbix() -> list[str]:
    return ["d", "e"]

# 类型检查工具会正确推断：adata为str，pdata为dict[str,str]，zdata为list[str]
adata, pdata, zdata = parallelize(get_airflow, get_prometheus, get_zabbix)

替代方案：使用executor.map #

如果所有Callable的返回值类型一致，map写法更简洁，但对于多类型返回值场景，上述方案的类型保留更精准：

def parallelize_map(*callables: Unpack[Tuple[Callable[[], Unpack[Ts]]]]) -> Tuple[Unpack[Ts]]:
    with ThreadPoolExecutor() as executor:
        return tuple(executor.map(lambda f: f(), callables))

兼容性说明 #

• 若需兼容Python 3.10以下版本，才需要手动编写多个@overload声明（如你最初尝试的方式），但这种方式最多只能覆盖有限数量的参数。
• 若Callable可能抛出异常，需在result()调用处添加异常处理，或在调用parallelize时统一捕获。

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