我有一个异步函数,它接受一个可以包含 CustomType1 或 CustomType2 值的列表作为输入。它还接受一个函数作为其第二个参数。现在我只使用
Callable[..., Coroutine[Any, Any, CustomType3]]
但我想正确输入函数参数。
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)))
我有两个具有不同签名的函数,可以传递给 func1:
async def fn1(
arg1: CustomType1,
arg2: str,
arg3: str
) -> CustomType3: ...
async def fn2(
arg1: CustomType2
) -> CustomType3: ...
我尝试过做
fn: Callable[[CustomType1 | CustomType2, ...], Coroutine[Any, Any, CustomType3]]
但是 mypy 会抛出这些错误:
Unexpected "..."
:在我输入 fn 参数的行Too few arguments
:在 func1 返回的行Argument 2 to "func1" has incompatible type "Callable[[CustomType1, str, str], Coroutine[Any, Any, CustomType3]]"; expected "Callable[[CustomType1 | CustomType2, Any], Coroutine[Any, Any, CustomType3]]"
:无论我在哪里调用 fn1,在调用 fn2 时都会出现类似的错误。我也尝试过https://mypy.readthedocs.io/en/stable/protocols.html#callback-protocols但我认为我有点超出了我的头脑。寻找有关如何解决此问题的任何指示。
(如果相关的话,我正在使用 mypy 版本 1.10.0)
要在 func1 中正确键入函数参数,您可以使用 Union 来处理 fn 的不同参数类型。定义自定义类型 CustomType1、CustomType2 和 CustomType3: `从输入导入 Any、Coroutine、Union、Callable、Protocol、TypeVar 导入异步
自定义类型1 = TypeVar('自定义类型1') CustomType2 = TypeVar('CustomType2') CustomType3 = TypeVar('CustomType3')`
创建一个与可能的函数签名相匹配的协议:
class FnProtocol(Protocol): def __call__(self, arg1: CustomType1, arg2: str, arg3: str) -> Coroutine[Any, Any, CustomType3]: ... def __call__(self, arg1: CustomType2) -> Coroutine[Any, Any, CustomType3]: ...
使用此协议在 func1 中键入 fn 参数:
async def func1( arg1: list[Union[CustomType1, CustomType2]], fn: FnProtocol, **kwargs: str, ) -> list[CustomType3]: return list(await asyncio.gather(*(fn(value, **kwargs) for value in arg1)))
定义示例函数 fn1 和 fn2:
` 异步 def fn1( arg1:自定义类型1, arg2: 字符串, arg3: 字符串 ) -> 自定义类型3: # 函数实现 ...
异步 def fn2( arg1:自定义类型2 ) -> 自定义类型3: # 函数实现 ... `