如何为asyncio.Task添加类型注释

Question

我有一个如下所示的代码：

import asyncio
from typing import List

def some_callback(result):
    print(result)

async def b() -> List[int]:
    return [1, 2, 3]

async def a() -> None:
    search = asyncio.ensure_future(b())
    search.add_done_callback( some_callback)
    await search

loop = asyncio.get_event_loop()
loop.run_until_complete(a())
loop.close()

我正在尝试向

some_callback

函数添加类型注释，但我无法完全理解如何注释

result

变量。应该是

Coroutine

吗？或者也许

Awaitable

？

当我使用

reveal_type

的

mypy

时，关于

result

变量的输出是

Any

。

该程序的输出是：

<Task finished coro=<b() done, defined at ____.py:7> result=[1, 2, 3]>

我应该如何正确记录这个函数？

Answer 1

通常，您只需打印变量的类型即可获得某些变量的基本注释：

def some_callback(result):
    print(type(result))

虽然它会表现出一些内在的

<class '_asyncio.Task'>

类型，但看起来我们可以将其视为常规

asyncio.Task

:

def some_callback(result):
    print(isinstance(result, asyncio.Task))  # True

但是正如您所指出的，我们还可以使用比

Task

更抽象的类型，例如

Awaitable

，因为

Task

是

Awaitable

（的子类）:

print(issubclass(asyncio.Task, typing.Awaitable))  # True

我们的选择现在缩小到

Task

或其父类之一，例如

Awaitable

（包括最极端的情况 -

Any

，它是任何类的父类，也是 mypy 向您建议的）。

add_done_callback

是

Future

的方法，根据文档将接收未来对象作为参数。它不会是任何类型的

Awaitable

（如协程），而只是

Future

或其某些子类，例如

Task

。

在选择类型注释时，对您的函数可以接受什么作为参数（正常工作）进行最抽象的描述以及对它可以返回什么进行最具体的描述是有意义的。因此，在

Future

和

Task

之间进行选择，我更喜欢

Future

（假设您不打算使用

Task

仅特定属性）。按照这个逻辑，最终答案是：

def some_callback(result: asyncio.Future):
    print(result)

这一切听起来有点复杂且耗时，但是一旦您有了想法，您将能够更快地选择注释。