Write Type Hints for Async Code

You are writing load, a function that takes a fetcher object and runs await fetcher.fetch(key). Your team has three fetchers. One uses async def. One is a cache that returns a result it already has. One starts a task and returns the task. All three work with await. You want one type hint that accepts all three, so mypy can check every caller.

What async def returns

import asyncio


async def fetch(key: str) -> bytes:
    await asyncio.sleep(0)
    return key.encode()


async def main() -> None:
    coro = fetch("a")  # a coroutine, nothing has run yet
    data = await coro  # b"a"

On an async def, the return annotation is the type you get after await, here bytes. Calling fetch("a") gives you a coroutine, and mypy writes its type as Coroutine[Any, Any, bytes]. The first two slots describe values that asyncio passes in and out while the coroutine is paused, so in your code they are Any. The last slot is the result.

Awaitable means you can await it

Awaitable[T], from collections.abc, is the type of anything you can await to get a T. Three kinds of object fit:

A coroutine, from calling an async def function.
A task, from asyncio.create_task.
A future. An asyncio.Future is a placeholder for a result that is not ready yet, like the futures from the thread pool articles. You await it to wait for the result. A task is a kind of future.

Every coroutine is awaitable, but a task or a future is not a coroutine. A coroutine has extra methods, such as close(), that the others lack. So accept Awaitable[T] when all you do is await the value. Accept Coroutine[Any, Any, T] when you need a real coroutine. For example, you might call close() on it, or pass it to asyncio.create_task, which accepts only coroutines.

Describe the fetcher with a protocol

A protocol is a class that lists the methods an object must have. Any class with matching methods fits the protocol, without inheriting from it. mypy checks the match wherever you pass an object in.

import asyncio
from collections.abc import Awaitable
from typing import Protocol


class Fetcher(Protocol):
    def fetch(self, key: str) -> Awaitable[bytes]: ...


class CacheFetcher:  # does not inherit from Fetcher
    def fetch(self, key: str) -> asyncio.Future[bytes]:
        future: asyncio.Future[bytes] = asyncio.get_running_loop().create_future()
        future.set_result(b"cached")
        return future


class NetworkFetcher:
    async def fetch(self, key: str) -> bytes:
        await asyncio.sleep(0.1)
        return key.encode()


async def load(fetcher: Fetcher, key: str) -> bytes:
    return await fetcher.fetch(key)

The ... body means the protocol describes the method without implementing it. Both classes fit it. CacheFetcher.fetch returns a future and NetworkFetcher.fetch returns a coroutine. Both are awaitable.

Common mistakes

The tempting way to write the protocol is the way you would write the method, with async def fetch(self, key: str) -> bytes. That says fetch returns a coroutine. A future is not a coroutine, so mypy rejects the cache:

error: Argument 1 to "load" has incompatible type "CacheFetcher"; expected "Fetcher"  [arg-type]
note:     Expected:
note:         def fetch(self, key: str) -> Coroutine[Any, Any, bytes]
note:     Got:
note:         def fetch(self, key: str) -> Future[bytes]

Write protocol methods with a plain def and an Awaitable[...] return type.

The opposite mistake is async def fetch(...) -> Awaitable[bytes] on a real method. The annotation is the type after await, so this says that awaiting fetch gives you yet another awaitable. Write -> bytes.

The third is forgetting await. The coroutine never runs, and nothing fails until much later. mypy reports an unused coroutine by default:

error: Value of type "Coroutine[Any, Any, bytes]" must be used  [unused-coroutine]
note: Are you missing an await?

To catch an unused task or future too, turn on mypy's optional unused-awaitable error code.

Check your understanding Your protocol says async def fetch(self, key: str) -> bytes. A teammate's class has def fetch(self, key: str) -> asyncio.Task[bytes]. Why does mypy reject it, and what do you change? Show answer

The async def in the protocol means fetch must return a coroutine. A task is awaitable, but it is not a coroutine, so the method doesn't match. Change the protocol to def fetch(self, key: str) -> Awaitable[bytes]. Coroutines, tasks and futures all fit, and await fetcher.fetch(key) still gives you bytes.

Going further

AsyncIterator[T] is the type of something you loop over with async for. An async def that contains yield returns one, and you can't await it. A later article in this track covers it.

A protocol with async def fetch accepts any method that returns a coroutine, including a plain def that returns one. mypy compares types, not the way you wrote the method.

See mypy's guide to typing async/await and its optional error codes.