How to add type annotation to asyncio.Task

Question

I have a code which look like this:

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()

I am trying to add type annotations to the some_callback function, but I can't fully understand hoe to annotate the result variable. Should it be Coroutine? Or maybe Awaitable?

When I am using the reveal_type of mypy, the output about the result variable is Any.

The output of this program is:

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

How should I document this function properly?

Answer 1

Usually you can get basic annotation for some variable with just printing it's type:

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

While it'll show some inner <class '_asyncio.Task'> type, looks like we can treat it as regular asyncio.Task:

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

But as you noted we can also use more abstract type then Task like Awaitable since Task is (subclass of) Awaitable:

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

Our choice narrowed down now to Task or one of its parent classes like Awaitable (including most extreme case - Any which is parent class for any class and which mypy've proposed to you).

add_done_callback is Future's method and according to documentation will receive future object as it's parameter. It won't be any kind of Awaitable (like coroutine), but only Future or some of it's subclasses like Task.

When it comes to choosing type annotation, it makes sense to be most abstract about what your function can accept as an argument (working properly) and most concrete about what it can return. So choosing between Future and Task I would prefer Future (assuming you aren't going to use Task specific only attrs). According to this logic, final answer is:

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

It all sounds a bit complecated and time consuming, but once you get an idea you'll be able to choose annotations much faster.