Typing async/await

Mypy supports the ability to type coroutines that use the async/awaitsyntax introduced in Python 3.5. For more information regarding coroutines andthis new syntax, see PEP 492.

Functions defined using async def are typed just like normal functions.The return type annotation should be the same as the type of the value youexpect to get back when await-ing the coroutine.

import asyncio
 
async def format_string(tag: str, count: int) -> str:
    return 'T-minus {} ({})'.format(count, tag)
 
async def countdown_1(tag: str, count: int) -> str:
    while count > 0:
        my_str = await format_string(tag, count)  # has type 'str'
        print(my_str)
        await asyncio.sleep(0.1)
        count -= 1
    return "Blastoff!"
 
loop = asyncio.get_event_loop()
loop.run_until_complete(countdown_1("Millennium Falcon", 5))
loop.close()

The result of calling an async def function without awaiting will be avalue of type Coroutine[Any, Any, T], which is a subtype ofAwaitable[T]:

my_coroutine = countdown_1("Millennium Falcon", 5)
reveal_type(my_coroutine)  # has type 'Coroutine[Any, Any, str]'

Note

reveal_type() displays the inferred static type ofan expression.

If you want to use coroutines in Python 3.4, which does not supportthe async def syntax, you can instead use the @asyncio.coroutinedecorator to convert a generator into a coroutine.

Note that we set the YieldType of the generator to be Any in thefollowing example. This is because the exact yield type is an implementationdetail of the coroutine runner (e.g. the asyncio event loop) and yourcoroutine shouldn’t have to know or care about what precisely that type is.

from typing import Any, Generator
import asyncio
 
@asyncio.coroutine
def countdown_2(tag: str, count: int) -> Generator[Any, None, str]:
    while count > 0:
        print('T-minus {} ({})'.format(count, tag))
        yield from asyncio.sleep(0.1)
        count -= 1
    return "Blastoff!"
 
loop = asyncio.get_event_loop()
loop.run_until_complete(countdown_2("USS Enterprise", 5))
loop.close()

As before, the result of calling a generator decorated with @asyncio.coroutinewill be a value of type Awaitable[T].

Note

At runtime, you are allowed to add the @asyncio.coroutine decorator toboth functions and generators. This is useful when you want to mark awork-in-progress function as a coroutine, but have not yet added yield oryield from statements:

import asyncio
 
@asyncio.coroutine
def serialize(obj: object) -> str:
    # todo: add yield/yield from to turn this into a generator
    return "placeholder"

However, mypy currently does not support converting functions intocoroutines. Support for this feature will be added in a future version, butfor now, you can manually force the function to be a generator by doingsomething like this:

from typing import Generator
import asyncio
 
@asyncio.coroutine
def serialize(obj: object) -> Generator[None, None, str]:
    # todo: add yield/yield from to turn this into a generator
    if False:
        yield
    return "placeholder"

You may also choose to create a subclass of Awaitable instead:

from typing import Any, Awaitable, Generator
import asyncio
 
class MyAwaitable(Awaitable[str]):
    def __init__(self, tag: str, count: int) -> None:
        self.tag = tag
        self.count = count
 
    def __await__(self) -> Generator[Any, None, str]:
        for i in range(n, 0, -1):
            print('T-minus {} ({})'.format(i, tag))
            yield from asyncio.sleep(0.1)
        return "Blastoff!"
 
def countdown_3(tag: str, count: int) -> Awaitable[str]:
    return MyAwaitable(tag, count)
 
loop = asyncio.get_event_loop()
loop.run_until_complete(countdown_3("Heart of Gold", 5))
loop.close()

To create an iterable coroutine, subclass AsyncIterator:

from typing import Optional, AsyncIterator
import asyncio
 
class arange(AsyncIterator[int]):
    def __init__(self, start: int, stop: int, step: int) -> None:
        self.start = start
        self.stop = stop
        self.step = step
        self.count = start - step
 
    def __aiter__(self) -> AsyncIterator[int]:
        return self
 
    async def __anext__(self) -> int:
        self.count += self.step
        if self.count == self.stop:
            raise StopAsyncIteration
        else:
            return self.count
 
async def countdown_4(tag: str, n: int) -> str:
    async for i in arange(n, 0, -1):
        print('T-minus {} ({})'.format(i, tag))
        await asyncio.sleep(0.1)
    return "Blastoff!"
 
loop = asyncio.get_event_loop()
loop.run_until_complete(countdown_4("Serenity", 5))
loop.close()

For a more concrete example, the mypy repo has a toy webcrawler thatdemonstrates how to work with coroutines. One versionuses async/awaitand oneuses yield from.