如何为接收Callable参数的异步函数正确标注类型？

问题背景 #

我有一个异步函数func1，第一个参数是包含CustomType1或CustomType2的列表，第二个参数是异步回调函数，返回CustomType3。当前用Callable[..., Coroutine[Any, Any, CustomType3]]做类型标注，但希望实现更精确的类型约束。

func1代码：

async def func1(
  arg1: list[CustomType1 | CustomType2],
  fn: Callable[..., Coroutine[Any, Any, CustomType3]],
  **kwargs: str,
) -> list[CustomType3]:
  return list(await asyncio.gather(*(fn(value, **kwargs) for value in arg1)))

可传入的两个回调函数：

async def fn1(
   arg1: CustomType1,
   arg2: str,
   arg3: str
) -> CustomType3: ...

async def fn2(
   arg1: CustomType2
) -> CustomType3: ...

尝试用Callable[[CustomType1 | CustomType2, ...], Coroutine[Any, Any, CustomType3]]标注时，mypy（1.10.0）抛出三个错误：

• 标注fn参数行出现Unexpected "..."
• func1返回值行出现Too few arguments
• 调用fn1/fn2时出现类型不兼容错误，例如Argument 2 to "func1" has incompatible type "Callable[[CustomType1, str, str], Coroutine[Any, Any, CustomType3]]"; expected "Callable[[CustomType1 | CustomType2, Any], Coroutine[Any, Any, CustomType3]]"

解决思路 #

普通Callable无法精准匹配两种不同的回调签名，需要用回调协议（Callback Protocols）结合函数重载来实现精确类型约束：

1. 定义对应回调签名的协议
   针对fn1和fn2的签名分别定义协议，明确参数类型和返回值：

   from typing import Protocol, Coroutine
   
   class Fn1Protocol(Protocol):
       async def __call__(self, arg1: CustomType1, arg2: str, arg3: str) -> CustomType3:
           ...
   
   class Fn2Protocol(Protocol):
       async def __call__(self, arg1: CustomType2) -> CustomType3:
           ...
   

2. 给func1添加重载声明
   通过重载明确arg1类型、回调类型、关键字参数的对应关系：

   from typing import overload, List
   
   @overload
   async def func1(
       arg1: list[CustomType1],
       fn: Fn1Protocol,
       *,
       arg2: str,
       arg3: str,
   ) -> list[CustomType3]:
       ...
   
   @overload
   async def func1(
       arg1: list[CustomType2],
       fn: Fn2Protocol,
   ) -> list[CustomType3]:
       ...
   
   # 实际实现保留原逻辑
   async def func1(
       arg1: list[CustomType1 | CustomType2],
       fn: Fn1Protocol | Fn2Protocol,
       **kwargs: str,
   ) -> list[CustomType3]:
       return list(await asyncio.gather(*(fn(value, **kwargs) for value in arg1)))
   

3. 关键说明

   • 普通Callable不支持参数列表中混合类型和...，协议则可以精准描述每个回调的完整签名。
   • 函数重载让mypy能识别：传入fn1时，arg1必须是纯CustomType1列表，且必须传arg2、arg3关键字参数；传入fn2时，arg1必须是纯CustomType2列表，不能有额外关键字参数。
   • 这样既满足了func1的通用逻辑，又能让类型检查工具识别出不匹配的调用场景。

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