asyncio deep dive — the model that runs modern Python services

Why this matters

Modern Python services — FastAPI APIs, LLM agents, websocket gateways, web scrapers — are async. Mastering asyncio (event loop, async def, await, TaskGroup, cancellation, async context managers, async generators) is the single biggest productivity multiplier for anything I/O-bound. This is a long lesson; you'll come back to it many times.

Learning objectives

1. Write async def functions and run them with asyncio.run.
2. Use gather, TaskGroup, as_completed, wait_for, wait.
3. Use Lock, Semaphore, Queue asynchronously.
4. Use async with, async for, async generators.
5. Handle cancellation, timeouts, and exception groups.

1. Mental model

A single thread runs an event loop. The loop runs coroutines cooperatively: each coroutine voluntarily yields control with await, and the loop resumes another ready coroutine.

When a coroutine awaits an I/O operation (await session.get(url)), it's suspended; the event loop registers a callback with the OS; when the OS reports the data is ready, the loop resumes the coroutine.

Trade-off:

• One process, one thread → no GIL contention; high concurrency.
• Any synchronous time.sleep(1) or CPU-bound loop blocks the entire loop. Never do that.

2. Hello, async

import asyncio

async def hello():
    print("hi")
    await asyncio.sleep(1)
    print("bye")

asyncio.run(hello())

• async def defines a coroutine function. Calling it returns a coroutine object (not running yet).
• asyncio.run(coro) creates an event loop, runs coro, returns its result.
• await coro pauses the current coroutine until coro completes.

You can await anything that's "awaitable": coroutines, Tasks, Futures. Synchronous functions are not awaitable.

3. Running things concurrently

asyncio.gather

Run many coroutines concurrently, collect results in input order:

async def fetch(url): ...

async def main():
    urls = ["https://a", "https://b", "https://c"]
    results = await asyncio.gather(*(fetch(u) for u in urls))

asyncio.run(main())

If one raises, by default gather cancels the others and re-raises. Use return_exceptions=True to collect exceptions as values:

results = await asyncio.gather(*coros, return_exceptions=True)
for r in results:
    if isinstance(r, Exception): log.warning(r)

asyncio.TaskGroup (3.11+) — the modern replacement

async def main():
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(fetch(url))
    # all tasks completed (or any failed -> whole group cancelled + ExceptionGroup raised)

Properties:

• Structured concurrency: tasks live within a scope.
• If any task fails, all are cancelled.
• Multiple failures become an ExceptionGroup (handle with except*).

Always prefer TaskGroup for new code. gather remains fine for "give me these results."

asyncio.as_completed — stream results as they finish

async def main():
    coros = [fetch(u) for u in urls]
    for fut in asyncio.as_completed(coros):
        result = await fut
        process(result)

asyncio.wait — fine-grained control

done, pending = await asyncio.wait(tasks, timeout=5, return_when=asyncio.FIRST_COMPLETED)
for p in pending: p.cancel()

Use when you need "first to finish" or timeouts on a set.

4. Tasks vs coroutines

A coroutine is passive — only runs when awaited. A Task is a coroutine scheduled on the event loop, running concurrently.

async def main():
    coro = work()                 # not running
    task = asyncio.create_task(work())   # scheduled; running
    await task                     # wait for completion

Use create_task (or TaskGroup.create_task) to start something "in the background."

Naming tasks (helps debugging)

task = asyncio.create_task(work(), name="worker-1")

asyncio.all_tasks() lists running tasks — useful in panic / shutdown.

5. Timeouts and cancellation

asyncio.timeout (3.11+)

async def main():
    try:
        async with asyncio.timeout(5):
            data = await slow_fetch(url)
    except TimeoutError:
        log.warning("timed out")

asyncio.timeout(seconds) is the modern, structured timeout. It cancels the wrapped block when it expires.

Older asyncio.wait_for(coro, timeout=...) still works but is harder to use correctly (cancellation semantics are fiddly).

Cancellation

task = asyncio.create_task(long())
await asyncio.sleep(1)
task.cancel()                     # request cancellation
try:
    await task
except asyncio.CancelledError:
    pass

Inside a coroutine, cancellation injects CancelledError at the next await. Never swallow CancelledError silently:

try:
    await something()
except asyncio.CancelledError:
    cleanup()
    raise                          # re-raise so the cancellation propagates

If you catch Exception broadly, also catch CancelledError separately (or BaseException, since CancelledError is no longer an Exception since 3.8 — it's BaseException).

try:
    await coro
except asyncio.CancelledError:
    cleanup()
    raise
except Exception as e:
    log.warning("error: %s", e)

6. Async context managers and generators

async with

async with httpx.AsyncClient() as client:
    r = await client.get(url)

Implements __aenter__ / __aexit__. Phase 3.3 covered the protocol.

@asynccontextmanager

from contextlib import asynccontextmanager

@asynccontextmanager
async def db_tx(pool):
    conn = await pool.acquire()
    try:
        async with conn.transaction():
            yield conn
    finally:
        await pool.release(conn)

async for + async generators

async def stream_chunks(url):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as r:
            async for chunk in r.aiter_bytes():
                yield chunk

async def main():
    async for chunk in stream_chunks(url):
        await save(chunk)

Built-in to libraries like httpx, aiofiles, aiokafka.

7. Sync primitives, async versions

import asyncio

lock = asyncio.Lock()
async with lock:
    do_thing()

sem = asyncio.Semaphore(10)
async with sem:                   # limit to 10 concurrent
    await fetch(url)

event = asyncio.Event()
await event.wait()
event.set(); event.clear()

q: asyncio.Queue[int] = asyncio.Queue(maxsize=100)
await q.put(1)
item = await q.get(); q.task_done()
await q.join()                    # wait until empty + all done

Semaphore is the canonical "rate-limit concurrent requests" tool:

async def fetch_all(urls):
    sem = asyncio.Semaphore(20)
    async def bound_fetch(u):
        async with sem:
            return await fetch(u)
    return await asyncio.gather(*(bound_fetch(u) for u in urls))

8. Mixing sync and async

Run a sync function in a thread

result = await asyncio.to_thread(blocking_function, arg1, arg2)

Useful when you must call a blocking library (legacy DB driver, image library). The function runs on a worker thread and the event loop continues.

For CPU-bound, use a process executor:

loop = asyncio.get_running_loop()
result = await loop.run_in_executor(process_pool, heavy_compute, x)

Call async from sync

asyncio.run(main())                    # top-level only

Inside a thread that doesn't own a loop, you can also use asyncio.run. Don't nest it inside an already-running loop — use await instead.

9. Async I/O libraries you'll actually use

For LLM work: openai, anthropic, google-genai SDKs all expose async clients.

10. Patterns

Bounded concurrency

async def fetch_all(urls, limit=20):
    sem = asyncio.Semaphore(limit)
    async def one(u):
        async with sem:
            return await fetch(u)
    return await asyncio.gather(*(one(u) for u in urls))

Fan-out, fan-in with TaskGroup

async def process(items):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(handle(item)) for item in items]
    return [t.result() for t in tasks]

Retry with exponential backoff

async def retry(coro_fn, *, attempts=3, base=0.1):
    for i in range(attempts):
        try:
            return await coro_fn()
        except (httpx.HTTPError, TimeoutError):
            if i + 1 == attempts: raise
            await asyncio.sleep(base * 2 ** i)

Periodic background task

async def heartbeat():
    while True:
        try:
            await send_heartbeat()
        except Exception as e:
            log.warning("heartbeat failed: %s", e)
        await asyncio.sleep(30)

asyncio.create_task(heartbeat(), name="heartbeat")

Graceful shutdown

import signal

async def main():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop.set)

    async with asyncio.TaskGroup() as tg:
        tg.create_task(server_loop(stop))
        tg.create_task(consumer_loop(stop))
        await stop.wait()
        # TaskGroup cancels remaining tasks on exit if they're still running

11. Common bugs

1. Forgetting await: calling foo() returns a coroutine object that does nothing. Most type checkers warn.
2. Calling a blocking function (time.sleep, requests.get, big NumPy op): freezes the loop.
3. Catching Exception and missing CancelledError: shutdown hangs.
4. Creating tasks but not awaiting them: silently lose results / errors. Use TaskGroup or keep a reference.
5. asyncio.run inside a running loop: RuntimeError. Use await instead.
6. Mixing event loops across threads.
7. Default thread pool exhaustion: to_thread shares a limited pool; many concurrent blocking calls queue.

12. Debugging

import asyncio
asyncio.run(main(), debug=True)     # warns on slow callbacks, missing awaits

Set PYTHONASYNCIODEBUG=1 env for the same effect.

Visualise with aiomonitor, aiodebug, or a tracing library (structlog, OpenTelemetry).

13. Worked example: async HTTP scraper

import asyncio
import httpx
from pathlib import Path

async def fetch(client: httpx.AsyncClient, url: str, sem: asyncio.Semaphore) -> tuple[str, bytes]:
    async with sem:
        r = await client.get(url, timeout=30)
        r.raise_for_status()
        return url, r.content

async def fetch_all(urls: list[str], *, limit: int = 20) -> dict[str, bytes]:
    sem = asyncio.Semaphore(limit)
    async with httpx.AsyncClient() as client:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, u, sem)) for u in urls]
        return dict(t.result() for t in tasks)

async def save(out: Path, data: dict[str, bytes]) -> None:
    out.mkdir(parents=True, exist_ok=True)
    for url, content in data.items():
        name = url.rsplit("/", 1)[-1] or "index"
        (out / name).write_bytes(content)

async def main(urls: list[str], out: Path) -> None:
    data = await fetch_all(urls)
    await save(out, data)

if __name__ == "__main__":
    urls = ["https://example.com/a", "https://example.com/b"]   # ...
    asyncio.run(main(urls, Path("downloads")))

20 concurrent requests, structured concurrency, clean shutdown. Compare to a threaded version — usually shorter and faster.

Hands-on lab (2.5 hours)

1. Write async def main(): that fetches 50 URLs concurrently with asyncio.gather + httpx.AsyncClient. Time it vs sequential.
2. Convert to TaskGroup. Test what happens when one URL 404s.
3. Add a Semaphore to cap concurrency at 10.
4. Replace gather(return_exceptions=True) semantics with TaskGroup + try/except* on ExceptionGroup.
5. Add asyncio.timeout(10) around the whole fetch; verify TimeoutError on a slow URL.
6. Write an async producer/consumer with asyncio.Queue; 1 producer pushing 100 items, 3 consumers processing.
7. Wrap a blocking function in await asyncio.to_thread(...); observe loop stays responsive.
8. Bonus: add graceful shutdown on Ctrl-C using loop.add_signal_handler.

Common pitfalls

1. Missing await (silent no-op).
2. Blocking the loop with sync I/O.
3. Forgetting to handle CancelledError.
4. Tasks created but not tracked → "Task was destroyed but it is pending!"
5. Calling asyncio.run inside an existing loop.
6. Mixing loops across threads.
7. Thinking async makes CPU work parallel.

Self-check

1. What is the event loop?
2. Difference between gather and TaskGroup.
3. How do you cancel a task?
4. How do you run a blocking function without blocking the loop?
5. What's an ExceptionGroup and how do you handle it?

References

• Python Concurrency with asyncio, Matthew Fowler.
• PEP 492 — Coroutines with async/await.
• PEP 530 — Asynchronous comprehensions.
• PEP 654 — Exception groups.
• PEP 657 — Better tracebacks for exceptions.
• Python docs, asyncio module.
• Yury Selivanov, "asyncio: lessons learned" (PyCon talk).