Python Advanced Patterns: asyncio + Type Hints

This guide combines two modern Python topics that show up together a lot in real codebases: asyncio (concurrency without threads) and static type hints with mypy. We'll focus on the patterns the language has settled on as of Python 3.11–3.12, noting which features need which minimum version.

Part 1 — asyncio

The event loop in one paragraph

Python's asyncio model is cooperative single-threaded concurrency: one OS thread runs a loop that picks one ready coroutine, runs it until it hits an await, then picks the next ready coroutine. Coroutines never pre-empt each other. That means asyncio is fantastic for I/O-bound work (network, disk, subprocesses) and useless for CPU-bound work — for the latter, use threads, processes, or native code.

Hello, async

import asyncio

async def greet(name: str) -> None:
    await asyncio.sleep(0.5)
    print(f"hello, {name}")

async def main() -> None:
    await greet("alex")

if __name__ == "__main__":
    asyncio.run(main())

A few rules that catch beginners:

• An async def function returns a coroutine object, not a result. Calling greet("alex") on its own does nothing — you must await it or schedule it.
• asyncio.run() creates the loop, runs the coroutine, and tears the loop down. Use it exactly once per program, at the top.
• You cannot call asyncio.run() from inside another running event loop (e.g. inside Jupyter — use await directly there).

Running things concurrently: gather vs TaskGroup

asyncio.gather() schedules multiple coroutines and waits for them all. It has been the workhorse for years, but it has rough edges around exception handling — if one task fails, the others keep running, and their results are lost unless you set return_exceptions=True.

Python 3.11 added asyncio.TaskGroup, which fixes those rough edges. If you're on 3.11+, prefer TaskGroup.

import asyncio

async def fetch(n: int) -> int:
    await asyncio.sleep(0.2)
    return n * n

async def main() -> None:
    results = await asyncio.gather(fetch(1), fetch(2), fetch(3))
    print(results)  # [1, 4, 9]

asyncio.run(main())

import asyncio

async def fetch(n: int) -> int:
    await asyncio.sleep(0.2)
    if n == 2:
        raise ValueError("no twos allowed")
    return n * n

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(fetch(1))
            t2 = tg.create_task(fetch(2))
            t3 = tg.create_task(fetch(3))
        # All tasks finished here. If any raised, *all* siblings are cancelled,
        # and the errors come back as an ExceptionGroup.
        print(t1.result(), t2.result(), t3.result())
    except* ValueError as eg:
        for err in eg.exceptions:
            print("caught:", err)

asyncio.run(main())

Note except* — that's the new "except-star" syntax for unpacking exception groups, also Python 3.11+. If you've never seen it before, it's a focused tool: it lets you catch and re-raise a subset of exceptions from an ExceptionGroup by type while letting the others propagate.

Fire-and-forget: the pitfall

asyncio.create_task() schedules a coroutine but does not wait for it. The classic mistake is not keeping a reference to the task:

# BUG: task may be garbage-collected mid-flight because no strong reference is held.
async def main() -> None:
    asyncio.create_task(slow_background_thing())
    await asyncio.sleep(10)

Always store the reference (in a set you manage, or on self), and ideally await it before the program exits:

background_tasks: set[asyncio.Task] = set()

async def main() -> None:
    task = asyncio.create_task(slow_background_thing())
    background_tasks.add(task)
    task.add_done_callback(background_tasks.discard)

    await asyncio.sleep(10)
    # On shutdown, await outstanding tasks so they can clean up.
    await asyncio.gather(*background_tasks, return_exceptions=True)

Cancellation

Tasks can be cancelled by calling task.cancel(). Inside the task, that delivers asyncio.CancelledError the next time it hits an await. You should let it propagate so cleanup runs — never swallow CancelledError without re-raising:

async def worker() -> None:
    try:
        while True:
            await asyncio.sleep(1)
            do_work()
    except asyncio.CancelledError:
        # Cleanup here.
        print("shutting down cleanly")
        raise   # always re-raise

async with, async for

Async context managers (__aenter__/__aexit__) and async iterators (__aiter__/__anext__) are how libraries expose resources whose acquisition/release does I/O. You'll meet them in aiohttp, asyncpg, etc.

import asyncio
import aiohttp

async def fetch(url: str) -> int:
    async with aiohttp.ClientSession() as session:    # async with
        async with session.get(url) as resp:
            return resp.status

async def main() -> None:
    urls = ["https://example.com", "https://example.org"]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    for t in tasks:
        print(t.result())

asyncio.run(main())

Calling blocking code from async

If a function is not async (e.g. requests.get, blocking file I/O, a CPU-bound function), calling it directly will block the whole event loop. Hand it off to a thread:

import asyncio
import requests   # synchronous library

async def fetch_blocking(url: str) -> int:
    # Runs in a default thread-pool executor, doesn't block the loop.
    resp = await asyncio.to_thread(requests.get, url, timeout=10)
    return resp.status_code

Part 2 — Type Hints & mypy

Type hints were introduced in PEP 484 for Python 3.5. They started as a static documentation tool and have since become a serious checker, with mypy as the most-used implementation. The hints themselves are stored in __annotations__ and are not enforced at runtime — Python doesn't check them; mypy does.

The modern flavour: PEP 585 + PEP 604

Pre-3.9 code looks like this:

from typing import List, Dict, Optional, Union

def lookup(ids: List[int], cache: Dict[int, str]) -> Optional[str]:
    ...

def fetch(url: Union[str, bytes]) -> None:
    ...

On Python 3.9+ you can use builtins as generics (PEP 585), and on 3.10+ you can use | for unions (PEP 604):

def lookup(ids: list[int], cache: dict[int, str]) -> str | None:
    ...

def fetch(url: str | bytes) -> None:
    ...

Both forms are valid; mix and match if you must, but settle on the modern form for new code.

TypedDict for dict shapes

Many APIs (especially JSON) pass dicts with fixed keys. TypedDict (PEP 589) gives them types:

from typing import TypedDict

class User(TypedDict):
    id: int
    name: str
    email: str | None      # nullable field

class UserWithOptionalAvatar(User, total=False):
    avatar_url: str        # may or may not be present

def render(u: User) -> str:
    return f"{u['name']} <{u['email']}>"

TypedDicts are excellent for typing JSON deserialisation. For full-blown data classes consider @dataclass or Pydantic instead.

Protocol — structural subtyping ("duck typing for type checkers")

Protocol (PEP 544) lets you say "I accept anything with this shape" without forcing the caller to inherit. This is essential for the duck-typed APIs Python is full of:

from typing import Protocol

class SupportsClose(Protocol):
    def close(self) -> None: ...

def safely(thing: SupportsClose) -> None:
    try:
        do_stuff(thing)
    finally:
        thing.close()

# Anything with a .close() method works — no inheritance needed.
safely(open("data.txt"))      # file objects
safely(some_db_connection)    # any DB connection with .close()

Final, Literal, NewType

from typing import Final, Literal, NewType

MAX_RETRIES: Final = 3              # mypy refuses re-assignment

Method = Literal["GET", "POST", "PUT", "DELETE"]

def request(url: str, method: Method = "GET") -> bytes:
    ...

UserId = NewType("UserId", int)      # nominal alias — int at runtime, distinct type to mypy

def load(uid: UserId) -> User:
    ...

load(42)              # mypy error: int is not UserId
load(UserId(42))      # OK

Generics

Before Python 3.12, generics used TypeVar:

from typing import TypeVar, Generic

T = TypeVar("T")

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []
    def push(self, item: T) -> None:
        self._items.append(item)
    def pop(self) -> T:
        return self._items.pop()

Python 3.12 added PEP 695, a much cleaner generic syntax — no more TypeVar imports:

class Stack[T]:
    def __init__(self) -> None:
        self._items: list[T] = []
    def push(self, item: T) -> None:
        self._items.append(item)
    def pop(self) -> T:
        return self._items.pop()

# Same for functions
def first[T](items: list[T]) -> T:
    return items[0]

Typing async code

The right hint for an async def function is whatever it returns — the async wrapping is implicit:

from collections.abc import Awaitable

async def fetch(url: str) -> bytes:        # returns bytes, not Coroutine[..., bytes]
    ...

def schedule(coro: Awaitable[bytes]) -> None:    # caller wants something awaitable
    ...

Running mypy

python -m pip install mypy

# Quick check
mypy myproject/

# Recommended for new code: strict mode (turns on a bundle of strict flags)
mypy --strict myproject/

# Individually useful flags if --strict is too much at first
mypy --disallow-untyped-defs --warn-unused-ignores myproject/

The --strict flag is the gold standard for new projects — it forces every function to be annotated, complains about implicit Any, and warns about dead type: ignore comments. For existing untyped code, start with no flags and ratchet up.

reveal_type — your debugging friend

Drop reveal_type(x) anywhere; mypy will print what type it inferred for x at that point. At runtime reveal_type is provided by mypy and ignored (or you need to import from typing on 3.11+).

from typing import reveal_type   # 3.11+, or just write it and mypy injects it

def example(x: list[int] | None) -> None:
    if x is None:
        return
    reveal_type(x)   # mypy: note: Revealed type is "builtins.list[builtins.int]"

Putting It Together: Typed Async Fetch

A small worked example that uses everything above:

"""Fetch a batch of URLs concurrently and return their statuses, fully typed."""
from __future__ import annotations

import asyncio
from typing import TypedDict
import aiohttp

class FetchResult(TypedDict):
    url: str
    status: int
    bytes: int

async def fetch_one(session: aiohttp.ClientSession, url: str) -> FetchResult:
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        body = await resp.read()
        return {"url": url, "status": resp.status, "bytes": len(body)}

async def fetch_all(urls: list[str]) -> list[FetchResult]:
    results: list[FetchResult] = []
    async with aiohttp.ClientSession() as session:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch_one(session, u)) for u in urls]
    # All tasks have completed (or the TaskGroup raised). Collect.
    for t in tasks:
        results.append(t.result())
    return results

def main() -> None:
    urls: list[str] = [
        "https://example.com",
        "https://example.org",
        "https://example.net",
    ]
    out = asyncio.run(fetch_all(urls))
    for r in out:
        print(f"{r['status']} {r['bytes']:7d}  {r['url']}")

if __name__ == "__main__":
    main()

Run mypy --strict typed_fetch.py — if you have it set up correctly, it returns zero errors. Then run python typed_fetch.py. You've now got concurrent I/O, structured concurrency with automatic cancellation on failure, and full static typing in fewer than 40 lines.

Further Reading

• PEP 484 — Type hints (the original)
• PEP 544 — Protocols (structural subtyping)
• PEP 585 — Type hinting generics in standard collections
• PEP 604 — Allow writing union types as X | Y
• PEP 654 — Exception Groups and except*
• PEP 695 — Type Parameter Syntax (the new 3.12 generic syntax)
• asyncio — official docs
• mypy documentation