Async Python with asyncio

Python’s asyncio runs many I/O operations concurrently on a single thread. While one coroutine waits on the network, the event loop runs another. Every example below was run with Python 3.13; aiohttp is 3.14.

Coroutines and async def

Calling an async def function does not run it — that surprises almost everyone the first time.

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)          # stand-in for a database query
    return {"id": user_id, "name": f"User-{user_id}"}

coro = fetch_user(42)
print(type(coro))
coro.close()                          # discard it without a warning

async def main() -> None:
    user = await fetch_user(42)
    print(user)

asyncio.run(main())

$ python coroutine.py
<class 'coroutine'>
{'id': 42, 'name': 'User-42'}

fetch_user(42) built a coroutine object and stopped. The body only ran once await drove it, inside the event loop that asyncio.run created. Each await is a point where the coroutine can pause and let other work run.

Running coroutines concurrently with gather

Three simulated requests take 0.3s, 0.5s and 0.2s. Run sequentially that is 1.0s; with gather it is the slowest one.

import asyncio
import time

async def fetch(url: str, delay: float) -> str:
    print(f"start {url}")
    await asyncio.sleep(delay)        # simulated network latency
    print(f"done  {url}")
    return f"response from {url}"

async def main() -> None:
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("/users", 0.3),
        fetch("/posts", 0.5),
        fetch("/comments", 0.2),
    )
    print(f"elapsed {time.perf_counter() - start:.1f}s")
    print(results)

asyncio.run(main())

$ python gather.py
start /users
start /posts
start /comments
done  /comments
done  /users
done  /posts
elapsed 0.5s
['response from /users', 'response from /posts', 'response from /comments']

All three start before any finishes, and they finish in order of delay — /comments first. Yet results comes back in argument order, not completion order, so you can unpack it positionally. The elapsed time (it varies by a few milliseconds run to run) is the maximum delay, not the sum.

Background work with create_task

create_task schedules a coroutine immediately and hands back a Task you can await later.

import asyncio

async def job(name: str, seconds: float) -> str:
    print(f"[{name}] started")
    await asyncio.sleep(seconds)
    print(f"[{name}] finished")
    return name

async def main() -> None:
    task_a = asyncio.create_task(job("A", 0.4))
    task_b = asyncio.create_task(job("B", 0.2))
    print("tasks created")          # runs before either task starts
    print(await task_a, await task_b)
    print(task_a.done(), task_b.done())

asyncio.run(main())

$ python tasks.py
tasks created
[A] started
[B] started
[B] finished
[A] finished
A B
True True

“tasks created” prints first: create_task only schedules the coroutine; it starts at main’s next await. Awaiting task_a first does not delay B — it was already running and finished while main waited on A.

Failing safely with TaskGroup

With gather, a failing task raises but its siblings keep running unattended. asyncio.TaskGroup (3.11+) cancels them instead.

import asyncio

async def worker(n: int) -> int:
    await asyncio.sleep(0.1 * n)
    if n == 2:
        raise ValueError(f"worker {n} failed")
    return n

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(worker(n)) for n in (1, 2, 3)]
    except* ValueError as eg:
        print("caught:", eg.exceptions)
    print([t.cancelled() for t in tasks])

asyncio.run(main())

$ python taskgroup.py
caught: (ValueError('worker 2 failed'),)
[False, False, True]

Worker 1 completed, worker 2 raised, and worker 3 — still sleeping — was cancelled the moment its sibling failed. Errors arrive wrapped in an ExceptionGroup, which is why the handler uses except*.

Real HTTP concurrency with aiohttp

requests blocks the thread; aiohttp does not. This example starts a tiny aiohttp server on localhost so it needs no internet (pip install aiohttp).

import asyncio
from aiohttp import web, ClientSession

async def handle(request: web.Request) -> web.Response:
    n = int(request.match_info["n"])
    await asyncio.sleep(0.1)                     # pretend each response is slow
    return web.json_response({"id": n, "title": f"post {n}"})

async def main() -> None:
    app = web.Application()
    app.add_routes([web.get("/posts/{n}", handle)])
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner, "127.0.0.1", 8081).start()

    async with ClientSession() as session:
        async def get(n: int) -> dict:
            async with session.get(f"http://127.0.0.1:8081/posts/{n}") as resp:
                resp.raise_for_status()
                return await resp.json()
        for post in await asyncio.gather(*(get(n) for n in (1, 2, 3))):
            print(post)
    await runner.cleanup()

asyncio.run(main())

$ python fetch_posts.py
{'id': 1, 'title': 'post 1'}
{'id': 2, 'title': 'post 2'}
{'id': 3, 'title': 'post 3'}

Server and client share one event loop: while the handlers sleep, the client’s three requests are all in flight. One ClientSession reuses a connection pool across the requests — create it once, not per call.

Async context managers

async with calls __aenter__ and __aexit__, which may themselves await — useful for connections whose open and close are network operations.

import asyncio

class Connection:
    async def __aenter__(self):
        await asyncio.sleep(0.05)     # simulated handshake
        print("connected")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        await asyncio.sleep(0.01)     # simulated flush
        print("closed, exception:", exc_type.__name__ if exc_type else None)

    async def query(self, sql: str) -> list[dict]:
        if "DROP" in sql:
            raise PermissionError("not allowed")
        return [{"id": 1}]

async def main() -> None:
    async with Connection() as db:
        print(await db.query("SELECT 1"))
    try:
        async with Connection() as db:
            await db.query("DROP TABLE users")
    except PermissionError as e:
        print("caller saw:", e)

asyncio.run(main())

$ python async_with.py
connected
[{'id': 1}]
closed, exception: None
connected
closed, exception: PermissionError
caller saw: not allowed

__aexit__ ran in both cases — including when the query raised — and it received the exception type. Because it returned None, the exception still propagated to the caller.

Timeouts

asyncio.timeout() (3.11+) puts a deadline on a block; asyncio.wait_for() does the same for one awaitable.

import asyncio

async def slow_operation() -> str:
    await asyncio.sleep(10)           # a server that never answers
    return "done"

async def main() -> None:
    try:
        async with asyncio.timeout(0.2):
            await slow_operation()
    except TimeoutError:
        print("timed out")

    result = await asyncio.wait_for(asyncio.sleep(0.05, result="fast"), timeout=1)
    print(result)

asyncio.run(main())

$ python timeout.py
timed out
fast

The 10-second sleep was cancelled after 0.2s, and the cancellation surfaced as the built-in TimeoutError (asyncio.TimeoutError is an alias for it since 3.11). An operation that beats its deadline returns normally.

asyncio patterns at a glance

Practice

1. Change gather.py to fetch the three URLs sequentially with three await lines, and confirm the elapsed time becomes the sum of the delays.

import asyncio
import time

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"response from {url}"

async def main() -> None:
    start = time.perf_counter()
    r1 = await fetch("/users", 0.3)
    r2 = await fetch("/posts", 0.5)
    r3 = await fetch("/comments", 0.2)
    print(f"elapsed {time.perf_counter() - start:.1f}s")
    print([r1, r2, r3])

asyncio.run(main())

$ python sequential.py
elapsed 1.0s
['response from /users', 'response from /posts', 'response from /comments']

2. Use asyncio.as_completed to print each result from gather.py as soon as it is ready instead of in argument order.

import asyncio

async def fetch(url: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"response from {url}"

async def main() -> None:
    coros = [fetch("/users", 0.3), fetch("/posts", 0.5), fetch("/comments", 0.2)]
    for next_done in asyncio.as_completed(coros):
        print(await next_done)

asyncio.run(main())

$ python as_completed.py
response from /comments
response from /users
response from /posts

3. Call gather on the three worker coroutines from taskgroup.py with return_exceptions=True, and print what it returns.

import asyncio

async def worker(n: int) -> int:
    await asyncio.sleep(0.1 * n)
    if n == 2:
        raise ValueError(f"worker {n} failed")
    return n

async def main() -> None:
    results = await asyncio.gather(*(worker(n) for n in (1, 2, 3)), return_exceptions=True)
    print(results)

asyncio.run(main())

$ python gather_errors.py
[1, ValueError('worker 2 failed'), 3]

4. Limit concurrency: run six 0.1-second jobs through an asyncio.Semaphore(2) and show that at most two run at once.

import asyncio

running = 0
peak = 0

async def job(sem: asyncio.Semaphore, n: int) -> None:
    global running, peak
    async with sem:
        running += 1
        peak = max(peak, running)
        await asyncio.sleep(0.1)
        running -= 1

async def main() -> None:
    sem = asyncio.Semaphore(2)
    await asyncio.gather(*(job(sem, n) for n in range(6)))
    print("peak concurrency:", peak)

asyncio.run(main())

$ python semaphore.py
peak concurrency: 2