How to add type annotation to asyncio.Task

I have a code that looks 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 some_callback function, but I cannot fully understand the hoe to annotate the result variable. Should it be Coroutine ? Or maybe Awaitable ?

When I use reveal_type of mypy , the output of the result variable is Any .

The output of this program:

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

How to properly document this feature?

+5
source share
1 answer

You can usually get a basic annotation for a variable by simply typing it:

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

While it will show some internal type <class '_asyncio.Task'> , it looks like we can treat it like a regular asyncio.Task :

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

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

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

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

add_done_callback Future method and according to the documentation will receive the future object as its parameter. It will not be any Awaitable (e.g. coroutine), but only Future or some of its subclasses, e.g. Task .

When it comes to choosing type annotations, it makes sense to be the most abstract about what your function can take as an argument (the right job) and the most specific about what it can return. Therefore, choosing between Future and Task , I would prefer Future (assuming that you are not going to use Task specific attrs only). According to this logic, the final answer is:

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

It all sounds a bit complete and time consuming, but as soon as you get an idea, you can choose annotations much faster.

+2
source

Source: https://habr.com/ru/post/1273550/


All Articles