Expert · Lesson 75 of 94

asyncio in practice

8 min readMahmudul Islam

Watch this lesson

View full playlist ↗

Read the lesson

What you'll learn

Run tasks concurrently and safely with asyncio.TaskGroup, and handle their errors as an ExceptionGroup
Bound waits with asyncio.timeout and understand cancellation
Write async generators and async context managers
Limit concurrency with a Semaphore so you do not overwhelm a service

Concept

gather is the quick way to run coroutines concurrently; asyncio.TaskGroup is the structured way. Inside async with asyncio.TaskGroup() as tg:, tg.create_task(coro()) starts a task, and the block does not exit until every task has finished. If one task raises, the others are cancelled and the exception surfaces after the block, so nothing runs on unnoticed. Several failures arrive together as an ExceptionGroup (3.11), which except* unpacks by type. Structured concurrency means a task never outlives the block that started it, which is what makes async programs debuggable.

asyncio in practice: async with TaskGroup() as tg: → tg.create_task(fetch(1)) → tg.create_task(fetch(2)) → tg.create_task(fetch(3)) → block exits only when all are done → others cancelled; ExceptionGroup raised after the block — asyncio in practice · Figure 1 · Open for full size

Waiting needs limits. async with asyncio.timeout(seconds): cancels whatever is inside if it runs too long and raises TimeoutError at the block. Cancellation in asyncio is cooperative: the task receives a CancelledError at its next await, and finally blocks and async context managers run, so resources are released. Catch CancelledError only to clean up, and always re-raise it, or the task will refuse to stop.

Tool	Job
`asyncio.TaskGroup`	run tasks concurrently; wait for all; propagate failures
`except* ValueError:`	handle one type out of an `ExceptionGroup`
`asyncio.timeout(s)`	cancel the block's work after s seconds
`asyncio.Semaphore(n)`	at most n tasks inside the block at once
`async for item in agen():`	consume an async generator
`async with resource:`	async context manager (`__aenter__` / `__aexit__`)
`asyncio.Queue`	producer/consumer between tasks
`asyncio.to_thread(f, ...)`	run blocking code without stalling the loop

Generators and context managers have async forms. An async generator uses async def with yield and may await between items, consumed with async for; it is how you stream pages from an API or lines from a socket one at a time. An async context manager defines __aenter__ and __aexit__, or uses @contextlib.asynccontextmanager, and is entered with async with; database connections, HTTP clients, and locks in async code all use it. The rules from the sync versions carry over: yield inside try/finally, cleanup in the exit.

Concurrency without a limit is a denial-of-service attack on whatever you are calling. asyncio.Semaphore(10) around the request means at most ten run at once while the rest wait their turn; the total still finishes far faster than sequentially, and the server stays happy. Combine it with a timeout per request and retries for transient failures, and you have the shape of every production async client.

asyncio in practice: 100 requests → Semaphore(10): ten at a time → asyncio.timeout(5) each → httpx.AsyncClient → results collected in a TaskGroup — asyncio in practice · Figure 2 · Open for full size

Code

A TaskGroup running fetches, one failing. The failure cancels the siblings and arrives as an ExceptionGroup, handled with except*.

# taskgroup.py
import asyncio


async def fetch(n):
    await asyncio.sleep(0.05 * n)
    if n == 3:
        raise ValueError(f"item {n} is bad")
    return n * 10


async def main():
    results = {}
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = {n: tg.create_task(fetch(n)) for n in range(1, 6)}
    except* ValueError as group:
        print("failures:", [str(e) for e in group.exceptions])
    for n, task in tasks.items():
        results[n] = "cancelled" if task.cancelled() else ("failed" if task.exception() else task.result())
    print(results)


asyncio.run(main())

Output

failures: ['item 3 is bad']
{1: 10, 2: 20, 3: 'failed', 4: 'cancelled', 5: 'cancelled'}

Timeouts and cooperative cancellation. The slow task is cancelled at its next await, its finally runs, and the timeout is reported cleanly.

# timeout_cancel.py
import asyncio


async def slow_job():
    try:
        await asyncio.sleep(5)
        return "finished"
    finally:
        print("slow_job cleanup ran")


async def main():
    try:
        async with asyncio.timeout(0.2):
            await slow_job()
    except TimeoutError:
        print("timed out after 0.2s")

    task = asyncio.create_task(slow_job())
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task cancelled:", task.cancelled())


asyncio.run(main())

Output

slow_job cleanup ran
timed out after 0.2s
slow_job cleanup ran
task cancelled: True

An async generator that pages through results with a wait between pages, and an async context manager built with asynccontextmanager.

# async_gen_ctx.py
import asyncio
import contextlib


async def pages(total, size):
    for start in range(0, total, size):
        await asyncio.sleep(0.01)                    # the network round trip
        yield list(range(start, min(start + size, total)))


@contextlib.asynccontextmanager
async def connection(name):
    await asyncio.sleep(0.01)
    print(f"open {name}")
    try:
        yield {"name": name, "queries": 0}
    finally:
        await asyncio.sleep(0.01)
        print(f"close {name}")


async def main():
    seen = []
    async for page in pages(7, 3):
        seen.append(page)
    print(seen)

    async with connection("db") as conn:
        conn["queries"] += 2
        print("inside with", conn)


asyncio.run(main())

Output

[[0, 1, 2], [3, 4, 5], [6]]
open db
inside with {'name': 'db', 'queries': 2}
close db

The tiny JSON server from the HTTP lesson, repeated here so this lesson runs on its own; it is written to disk and not run directly.

# server.py
import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

USERS = {1: {"id": 1, "name": "ada", "active": True}, 2: {"id": 2, "name": "linus", "active": False}}


class Handler(BaseHTTPRequestHandler):
    def log_message(self, *args):
        pass

    def send_json(self, status, payload):
        body = json.dumps(payload).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        url = urlparse(self.path)
        if url.path == "/users":
            active = parse_qs(url.query).get("active")
            users = list(USERS.values())
            if active:
                users = [u for u in users if u["active"] == (active[0] == "true")]
            return self.send_json(200, users)
        if url.path.startswith("/users/"):
            user = USERS.get(int(url.path.rsplit("/", 1)[1]))
            return self.send_json(200, user) if user else self.send_json(404, {"error": "no such user"})
        if url.path == "/delay":
            time.sleep(0.2)
            return self.send_json(200, {"waited": 0.2})
        if url.path == "/slow":
            time.sleep(2)
            return self.send_json(200, {})
        if url.path == "/boom":
            return self.send_json(503, {"error": "try later"})
        self.send_json(404, {"error": "unknown path"})

    def do_POST(self):
        length = int(self.headers.get("Content-Length", 0))
        data = json.loads(self.rfile.read(length) or b"{}")
        if "name" not in data:
            return self.send_json(400, {"error": "name is required"})
        new_id = max(USERS) + 1
        USERS[new_id] = {"id": new_id, "name": data["name"], "active": True}
        self.send_json(201, USERS[new_id])


class Server(ThreadingHTTPServer):
    request_queue_size = 32          # ten connections arriving at once must not overflow the backlog


def start_server():
    server = Server(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    return f"http://127.0.0.1:{server.server_port}"

Code walkthrough

Line 7: Two users in a dictionary that acts as the database.
Line 10-12: Silence the request log so outputs stay clean.
Line 14-20: A helper that writes a JSON response with the right headers.
Line 22-29: GET on slash users, with an optional active filter from the query string.
Line 30-32: GET one user by id, or a 404.
Line 33-41: A delay path that waits a fifth of a second, a slow path for timeouts, a boom path that returns 503, and a catch-all 404.
Line 43-50: POST reads the body, requires a name, and creates a user with status 201.
Line 53-54: A threaded server whose listen backlog is large enough for ten connections arriving at once; the default of five would leave the rest waiting on TCP retries.
Line 57-60: Start it on a free port in a background thread and return its base URL.

The production shape: bounded concurrency with a Semaphore, a per-request timeout, and a TaskGroup, against the local server. Thirty requests, at most five in flight.

# bounded_client.py
import asyncio
import time

import httpx

from server import start_server

in_flight = {"now": 0, "peak": 0}


async def fetch(client, sem, user_id):
    async with sem:
        in_flight["now"] += 1
        in_flight["peak"] = max(in_flight["peak"], in_flight["now"])
        try:
            async with asyncio.timeout(2):
                response = await client.get(f"/users/{user_id}")
            await asyncio.sleep(0.02)
            return response.status_code
        finally:
            in_flight["now"] -= 1


async def main(base):
    sem = asyncio.Semaphore(5)
    async with httpx.AsyncClient(base_url=base, timeout=5) as client:
        start = time.perf_counter()
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(fetch(client, sem, (i % 3) + 1)) for i in range(30)]
        codes = [t.result() for t in tasks]
    print(sorted(set(codes)), len(codes))
    print("peak concurrency:", in_flight["peak"], "| under 1s:", time.perf_counter() - start < 1.0)


asyncio.run(main(start_server()))

Output

[200, 404] 30
peak concurrency: 5 | under 1s: True

Try it yourself

Build crawler.py: an async function that fetches /users/<id> for ids 1 to 50 from the local server with a Semaphore(8), a two-second timeout per request, and a TaskGroup, collecting status codes into a Counter. Make the server's /slow route part of the id set so one request times out, and handle it so the others still complete. Then write an async generator that yields the results as they finish, using asyncio.as_completed, and consume it with async for.

Common mistakes

Swallowing CancelledError. Catching it without re-raising keeps a cancelled task alive; clean up in finally and let it propagate.
Unbounded fan-out. gather over ten thousand requests opens ten thousand connections; put a Semaphore around the work.
gather when a failure should stop everything. It keeps running the others by default; TaskGroup cancels siblings and surfaces the error.

What's next

TaskGroup runs tasks in a structured block; failures cancel siblings and surface as an ExceptionGroup handled with except*.
asyncio.timeout bounds waits; cancellation is cooperative and runs finally blocks; always re-raise CancelledError.
Async generators and context managers stream and manage resources; a Semaphore bounds concurrency.

You can write real async services. Next you step back to the question under all performance work: how does the cost of code grow with its input?

Course overview