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

How to add type annotation to asyncio.Task

我的代码如下所示:

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?

当我使用 mypyreveal_type 时,关于 result 变量的输出是 Any.

这个程序的输出是:

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

我应该如何正确记录此函数?

通常你可以通过打印它的类型来获得一些变量的基本注释:

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

虽然它会显示一些内部 <class '_asyncio.Task'> 类型,但看起来我们可以将其视为常规 asyncio.Task:

def some_callback(result):
    print(type(result) is asyncio.Task)  # True

但是正如您所指出的,我们还可以使用比 Task 更抽象的类型,例如 Awaitable,因为 TaskAwaitable 的(subclass):

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

我们的选择现在缩小到 Task 或其父 class 之一,例如 Awaitable(包括最极端的情况 - Any 是父 class 对于任何 class 以及 mypy 向您推荐的)。

add_done_callbackFuturemethod 并且根据文档将接收未来对象作为其参数。它不会是任何类型的 Awaitable(如协程),而只是 Future 或其中的一些子 class,例如 Task.

在选择类型注释时,对于您的函数可以接受什么作为参数(正常工作)最抽象和它可以接受什么最具体是有意义的 return。因此,在 FutureTask 之间进行选择,我更喜欢 Future (假设您不打算使用 Task 特定的属性)。按照这个逻辑,最终答案是:

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

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