Asynchronous Programming with asyncio

Advanced⏱90 minutes📖13 min readLesson 2 of 10

Asynchronous Programming with asyncio

Reason about the event loop, structure concurrent work with TaskGroup, cancel and time out cleanly, and debug a stuck service with the tools Python 3.14 and 3.15 added

A service handles forty requests per second on a laptop and eight on the production host. The handler awaits a database call, then parses a JSON payload with a synchronous library, then awaits a webhook. Nothing looks wrong in code review. The problem is a single blocking call: while the parser runs, the event loop cannot schedule any other task, so every other request waits behind it.

async and await describe a style of cooperation, not a performance setting. A coroutine yields control at an await, and the loop runs another ready task. If a coroutine never yields, or yields to something that blocks the thread, the whole process becomes sequential. Most asyncio bugs are this one bug wearing different clothes.

This lesson builds the model first: one thread, one loop, a queue of ready callbacks, and a selector that reports when sockets and timers are ready. On top of that model it covers tasks, TaskGroup, cancellation and timeouts, synchronization primitives, the bridge to blocking libraries, streams, HTTP client limits, and the introspection commands added in Python 3.14. The closing sections measure a concurrency limit and walk through three production systems.

ℹ️Note

The examples run on Python 3.14 or newer. TaskGroup.cancel requires 3.15, and python -m asyncio ps requires 3.14. Everything else works on 3.11 and later.

Key terms

Event loop

A single-threaded scheduler that runs ready callbacks, polls the operating system for I/O readiness and resumes waiting coroutines.

Coroutine

A function defined with async def. Calling it returns a coroutine object that must be awaited or scheduled.

Task

A coroutine scheduled on the loop, with its own state, name and result.

Future

A low-level placeholder for a result that will be set later by the loop or by a callback.

Structured concurrency

A discipline in which concurrent work starts and finishes inside one lexical scope, so no task outlives its parent.

TaskGroup

The standard structured concurrency primitive. It waits for all child tasks and combines their failures into an ExceptionGroup.

Cancellation

A request to stop a task by raising CancelledError inside it at the next await point.

Timeout

A deadline attached to a block of awaits, implemented with asyncio.timeout and raised as TimeoutError.

Backpressure

A limit that stops a producer from outrunning a consumer, expressed as a bounded queue, a semaphore or a socket write buffer.

GIL-bound bridge

A call from async code into a blocking C function that holds the interpreter and prevents the loop from running.

Selector

The operating system interface that reports which file descriptors are ready for reading or writing.

Eager task factory

A task factory that starts a coroutine immediately instead of scheduling its first step, reducing latency for short tasks.

1. One Thread and a Queue of Ready Work

The event loop runs one callback at a time to completion. When a coroutine awaits, it suspends and hands control back to the loop, which picks the next ready callback. When a socket becomes readable, the selector reports it and the loop resumes the coroutine that was waiting on it.¹

That design makes every blocking call a global problem. time.sleep blocks the only thread, so no other task runs. requests.get does the same. A CPU-heavy parser does the same. The loop has no way to preempt a callback that refuses to yield.

python

import asyncio
import time

async def blocking() -> None:
    print("blocking starts")
    time.sleep(1)
    print("blocking ends")

async def ticker() -> None:
    for index in range(5):
        print("tick", index)
        await asyncio.sleep(0.2)

async def main() -> None:
    await asyncio.gather(blocking(), ticker())

asyncio.run(main())

All five ticks appear after the one-second sleep. Replace time.sleep(1) with await asyncio.sleep(1) and the ticks interleave, because the await yields control while the timer is pending.

asyncio.run creates a loop, runs one coroutine to completion and closes the loop. It cannot be called from a running loop, and it should not be called more than once in a process. For a long-lived service, the framework owns the loop and calls your coroutine as a request handler.²

2. Coroutines, Tasks and Futures

A coroutine object does nothing until it is awaited or wrapped in a task. asyncio.create_task schedules it immediately and returns a Task you can await, cancel or inspect.³

python

import asyncio

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

async def main() -> None:
    first = asyncio.create_task(fetch("fast", 0.1), name="fast")
    second = asyncio.create_task(fetch("slow", 0.3), name="slow")
    print(first.get_name(), second.get_name())
    print(await first)
    print(await second)

asyncio.run(main())

Names show up in tracebacks and in the introspection commands later in this lesson, so name every long-lived task. Keep a reference to each task you create. The loop holds only a weak reference, and a task with no other reference can be garbage collected while it is still pending.

A Future is the lower-level object underneath a task. Library code creates futures and resolves them from callbacks. Application code should almost always use tasks, because a bare future has no coroutine attached and no cancellation semantics of its own.

python

import asyncio

async def main() -> None:
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    loop.call_later(0.2, future.set_result, 42)
    print(await future)

asyncio.run(main())

Python 3.12 added asyncio.eager_task_factory. With it, create_task runs the coroutine's first step synchronously instead of scheduling it. Short tasks that complete without awaiting finish immediately, which removes a loop iteration of latency.⁴ The tradeoff is that the caller now pays for that first step, so a task that does real work before its first await delays the caller.

python

import asyncio

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.set_task_factory(asyncio.eager_task_factory)
    task = asyncio.create_task(asyncio.sleep(0, result="done"))
    print(task.done())
    print(await task)

asyncio.run(main())

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Description

A class diagram of the main asyncio objects, with Runner creating the loop, Task extending Future, TaskGroup owning child tasks, and Timeout cancelling the work inside a block.

3. Structured Concurrency with TaskGroup

asyncio.gather starts a set of awaitables and returns their results, but it does not own their lifetime in a strong sense. If one child fails, the others keep running unless you cancel them yourself. TaskGroup, added in 3.11, makes the group a scope: the async with block does not exit until every child has finished, and a failure cancels the remaining children.⁵

python

import asyncio

async def job(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    if name == "bad":
        raise ValueError(f"{name} failed")
    return f"{name} ok"

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(job("one", 0.1))
            group.create_task(job("bad", 0.2))
            group.create_task(job("three", 0.3))
    except* ValueError as errors:
        print("failed:", [str(error) for error in errors.exceptions])

asyncio.run(main())

The except* syntax handles the ExceptionGroup that the group raises. Each child failure becomes a member, and you can match on the type. The cancelled siblings do not appear as failures, because cancellation from the group is not an error.

Python 3.15 adds TaskGroup.cancel(), which cancels every child without raising an artificial exception to break out of the block. That pattern used to require a sentinel task or a custom exception.⁶

python

import asyncio

async def poll(stop: asyncio.Event) -> None:
    while not stop.is_set():
        await asyncio.sleep(0.05)

async def main() -> None:
    stop = asyncio.Event()
    async with asyncio.TaskGroup() as group:
        group.create_task(poll(stop))
        await asyncio.sleep(0.12)
        stop.set()
        group.cancel()

asyncio.run(main())

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Description

A sequence diagram of two tasks sharing one event loop, where each task runs until an await registers a socket or timer, the selector reports readiness, and the loop resumes the waiting task.

gather is still the right tool for a simple fan-out where every result is required and failures should be inspected together. Use return_exceptions=True when a partial result is better than no result, and inspect each entry.

python

import asyncio

async def might_fail(value: int) -> int:
    if value == 2:
        raise ValueError("two")
    return value

async def main() -> None:
    results = await asyncio.gather(
        *(might_fail(value) for value in range(4)),
        return_exceptions=True,
    )
    print(results)

asyncio.run(main())

The output contains three integers and one ValueError. That is a valid choice for a report that can tolerate missing rows, and a bad choice for a payment batch where a missing row is a bug.

The video below builds a small web crawler with async and await, which is the same fan-out pattern the task group examples use at a smaller scale.

mCodingWatch on YouTube

Description

A video lesson that introduces async and await, explains how the event loop interleaves waiting work, and builds a web crawler that fetches several pages concurrently.

4. Cancellation, Cleanup and Timeouts

Cancellation is cooperative. task.cancel() requests it, and the loop raises CancelledError inside the task at its next await. If the coroutine catches Exception, it also catches CancelledError in older code, because CancelledError used to inherit from Exception. Since 3.8 it inherits from BaseException, so a normal except Exception no longer swallows it.

python

import asyncio

async def worker() -> None:
    try:
        await asyncio.sleep(5)
    except asyncio.CancelledError:
        print("cleaning up")
        raise

async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(0.1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled")

asyncio.run(main())

Always re-raise CancelledError after cleanup. Swallowing it makes the task look successful and breaks the cancellation contract that TaskGroup and timeouts depend on.

asyncio.shield protects an inner awaitable from cancellation of the outer task. It is useful when a write must finish even if the caller goes away, and it is dangerous when it hides a hung operation. The shielded task keeps running after the outer task is cancelled, so somebody must still own it.

python

import asyncio

async def commit() -> None:
    await asyncio.sleep(0.2)
    print("committed")

async def handler() -> None:
    try:
        await asyncio.shield(commit())
    except asyncio.CancelledError:
        print("caller left, commit continues")
        raise

async def main() -> None:
    task = asyncio.create_task(handler())
    await asyncio.sleep(0.05)
    task.cancel()
    await asyncio.sleep(0.3)

asyncio.run(main())

asyncio.timeout attaches a deadline to a block. It raises TimeoutError at the point where the block is waiting, cancels the inner task and leaves the outer task running.³ Prefer it to wait_for, which wraps a single awaitable and is easy to apply at the wrong scope.

python

import asyncio

async def slow() -> None:
    await asyncio.sleep(2)

async def main() -> None:
    try:
        async with asyncio.timeout(0.2):
            await slow()
    except TimeoutError:
        print("deadline reached")

asyncio.run(main())

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Description

A task state diagram showing a task move from pending to running, then to done, cancelled or failed, with cancellation able to arrive while the task is pending or running.

Timeouts compose with TaskGroup. A timeout around the whole group bounds the total work, and a timeout inside each child bounds one attempt. Mixing the two without thought produces a group that cancels healthy siblings because one child was slow.

python

import asyncio

async def attempt(name: str, delay: float) -> str:
    async with asyncio.timeout(0.15):
        await asyncio.sleep(delay)
    return name

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(attempt("fast", 0.05))
            group.create_task(attempt("slow", 0.5))
    except* TimeoutError:
        print("one attempt timed out")

asyncio.run(main())

5. Locks, Semaphores and Queues

Async primitives look like their thread counterparts but they never block the thread. They suspend the task and let the loop run something else. Never share an asyncio.Lock with a thread, and never use a threading.Lock inside a coroutine, because a contended thread lock blocks the loop.⁷

Primitive	Use
`Lock`	Protect a critical section that must not interleave
`Semaphore`	Limit concurrent access to a resource
`Event`	Signal a state change to many waiters
`Condition`	Wait for a predicate, with notification
`Queue`	Hand work between producer and consumer tasks

A semaphore is the simplest way to bound concurrency. The code below allows ten requests at a time and leaves the rest waiting, which keeps memory and connection count bounded.

python

import asyncio

async def fetch(index: int, semaphore: asyncio.Semaphore) -> int:
    async with semaphore:
        await asyncio.sleep(0.1)
        return index

async def main() -> None:
    semaphore = asyncio.Semaphore(10)
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch(index, semaphore)) for index in range(100)]
    print(len(tasks))

asyncio.run(main())

asyncio.Queue is the better tool when producer and consumer rates differ. The maxsize argument provides backpressure: put waits when the queue is full, so a fast producer cannot grow memory without bound.⁸

python

import asyncio

async def producer(queue: asyncio.Queue[int | None]) -> None:
    for value in range(5):
        await queue.put(value)
    await queue.put(None)

async def consumer(queue: asyncio.Queue[int | None]) -> None:
    while True:
        value = await queue.get()
        try:
            if value is None:
                return
            print("got", value)
        finally:
            queue.task_done()

async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=2)
    await asyncio.gather(producer(queue), consumer(queue))

asyncio.run(main())

Python 3.13 added Queue.shutdown(). After shutdown, get raises asyncio.QueueShutDown instead of waiting forever, which gives a clean way to stop consumers during application shutdown.⁴

python

import asyncio

async def main() -> None:
    queue: asyncio.Queue[int] = asyncio.Queue()
    queue.shutdown()
    try:
        await queue.get()
    except asyncio.QueueShutDown:
        print("queue closed")

asyncio.run(main())

Matching Question

How to answer

Matching exercise with 5 pairs. Match each item on the left with its answer on the right.

Match each asyncio primitive to the problem it solves.

Semaphore

Drop here

Queue with maxsize

Drop here

Event

Drop here

Lock

Drop here

TaskGroup

Drop here

Drag answers from here:

Limit how many tasks use a resource at once

Apply backpressure between producer and consumer

Wake every waiter when a state changes

Keep a critical section from interleaving

Tie the lifetime of child tasks to one scope

6. Bridging Blocking Code

Some libraries have no async API. Database drivers, image codecs and legacy SDKs block the calling thread. Run them in a worker thread with asyncio.to_thread, which wraps run_in_executor and propagates context variables.³

python

import asyncio
import time

def parse_document(text: str) -> int:
    time.sleep(0.2)
    return len(text.split())

async def main() -> None:
    text = "one two three four"
    words = await asyncio.to_thread(parse_document, text)
    print(words)

asyncio.run(main())

to_thread uses the loop's default executor, which is a thread pool. That pool shares the GIL, so it helps with blocking I/O and not with pure Python CPU work. For CPU work, use a process pool and loop.run_in_executor with a ProcessPoolExecutor, or move the work to a separate service.

AnyIO provides a backend-neutral version, anyio.to_thread.run_sync, with abandon_on_cancel to decide whether a cancelled call keeps running in the thread.⁹ Trio offers the same idea with its own API.¹⁰ Choose one of these when you want code that runs on more than one async framework.

plaintext

import anyio
import time

def parse_document(text: str) -> int:
    time.sleep(0.2)
    return len(text.split())

async def main() -> None:
    words = await anyio.to_thread.run_sync(parse_document, "one two three")
    print(words)

anyio.run(main)

Do not call to_thread for everything. A thread per blocking call is still a thread, with a stack and a context switch. Batch small calls, cache results, and measure whether the thread hop costs more than the work.

7. Streams and Backpressure

asyncio.start_server and asyncio.open_connection expose TCP as a reader and a writer. reader.read(n) waits for bytes, and writer.drain() waits until the transport buffer has room.¹¹ The drain call is the backpressure signal: a producer that never awaits it can buffer the whole response in memory.

plaintext

import asyncio

async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    while data := await reader.read(1024):
        writer.write(data.upper())
        await writer.drain()
    writer.close()
    await writer.wait_closed()

async def main() -> None:
    server = await asyncio.start_server(handle, "127.0.0.1", 8888)
    async with server:
        await server.serve_forever()

asyncio.run(main())

start_server accepts a limit argument for the reader buffer. A small limit keeps memory bounded and forces the peer to slow down. A large limit improves throughput for big payloads and increases the memory a slow consumer can pin.

python

import asyncio

async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    data = await reader.readline()
    writer.write(data[::-1])
    await writer.drain()
    writer.close()
    await writer.wait_closed()

async def main() -> None:
    server = await asyncio.start_server(handle, "127.0.0.1", 0)
    port = server.sockets[0].getsockname()[1]
    async with server:
        reader, writer = await asyncio.open_connection("127.0.0.1", port)
        writer.write(b"hello\n")
        await writer.drain()
        print(await reader.readline())
        writer.close()
        await writer.wait_closed()

asyncio.run(main())

The second example prints b"olleh\n". It runs in one process and one loop, which makes it a useful smoke test for the stream API without a network dependency.

8. HTTP Clients, Timeouts and Connection Limits

An async HTTP client is only useful if it reuses connections. Creating a client per request pays for a TCP handshake and a TLS handshake every time, and it defeats the pool limits. Create one client for the lifetime of the application and close it during shutdown.

HTTPX 0.28.1 was the last release of the original package. Pydantic maintains a fork named httpx2, and the OpenAI and Anthropic SDKs depend on it. The import name is httpx2, the API is the same, and the package requires Python 3.11 or newer.¹²

plaintext

import httpx2

timeout = httpx2.Timeout(connect=2.0, read=5.0, write=5.0, pool=1.0)
limits = httpx2.Limits(max_connections=20, max_keepalive_connections=10)

async def fetch(client: httpx2.AsyncClient, url: str) -> str:
    response = await client.get(url)
    response.raise_for_status()
    return response.text

async def main() -> None:
    async with httpx2.AsyncClient(timeout=timeout, limits=limits) as client:
        print(await fetch(client, "https://example.com"))

asyncio.run(main())

The four timeout values cover different waits. connect bounds the handshake, read bounds the wait for a response byte, write bounds sending the request, and pool bounds the wait for a free connection. A single scalar timeout hides which one fired.

aiohttp uses a connector for the same job. TCPConnector(limit=20) caps total connections and limit_per_host caps one host. Both clients raise a timeout error that you should map to a 504 at the API boundary rather than letting it become an unhandled exception.

plaintext

import aiohttp

async def main() -> None:
    connector = aiohttp.TCPConnector(limit=20, limit_per_host=5)
    timeout = aiohttp.ClientTimeout(total=10, connect=2)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        async with session.get("https://example.com") as response:
            print(response.status, await response.text())

asyncio.run(main())

Retries need a policy, not a loop. Retry only idempotent methods, cap the number of attempts, add jitter, and respect Retry-After. A retry inside a timeout can double the time a caller waits, so the outer deadline must include the retries.

9. Debugging a Running Loop

A hung asyncio service is hard to inspect with a normal stack dump, because the stack shows the loop waiting in the selector, not the task that is stuck. Python 3.14 added a command-line inspector that reads another process and prints its tasks.¹³

bash

python -m asyncio ps 12345
python -m asyncio pstree 12345

ps prints a flat table of tasks, their names, their coroutine stacks and which task awaits them. pstree renders the same data as a tree, which shows the parent and child relationships in a TaskGroup.

Turn on debug mode during development. It enables slow callback warnings and extra checks around thread safety. asyncio.run(main(), debug=True) is enough for a script, and PYTHONASYNCIODEBUG=1 covers a service started by a framework such as Uvicorn.¹⁴

python

import asyncio

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.slow_callback_duration = 0.05
    await asyncio.sleep(0)

asyncio.run(main(), debug=True)

When a task is stuck, print asyncio.all_tasks() with names and stacks. When the loop itself is starved, the problem is a blocking callback, and a sampling profiler is the right tool because it shows where the thread spends its time.

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Description

A flowchart of the cancellation path from a running task through a cancel request, cleanup in finally, and either a re-raised cancellation or a swallowed one that breaks structured concurrency.

Ordering Question

How to answer

Ordering exercise with 5 items. Move them into the correct order, then check your answer.

Order one iteration of the event loop.

Run callbacks that are already ready

Run the first step of newly scheduled tasks

Poll the selector for ready sockets and expired timers

Resume the coroutines whose events are ready

Schedule the next iteration

10. What Changed in 2026

Change	Version	Why it matters
`python -m asyncio ps PID` and `pstree PID`	3.14	Inspect tasks in a running process without restarting it
`TaskGroup.cancel()`	3.15	Cancel a group without a sentinel exception
`Queue.shutdown()` and `QueueShutDown`	3.13	Stop consumers cleanly during shutdown
`eager_task_factory`	3.12	Run short tasks immediately and cut a loop iteration
`asyncio.timeout` and `TaskGroup`	3.11	Structured concurrency and scoped deadlines
`httpx2` becomes the maintained HTTPX line	2026	New async clients should use the fork, and SDKs already do

The older asyncio.wait_for still works, and gather is still useful for simple fan-out. The change is what to teach first: a group with a scope and a timeout with a block prevent more bugs than a bag of loose tasks.

💡 Tip: Use Ctrl + Mouse Wheel to zoom, or click the fullscreen icon

Description

A timeline of asyncio features from task groups and timeouts in Python 3.11, through the eager task factory in 3.12 and queue shutdown in 3.13, to process introspection in 3.14 and TaskGroup.cancel in 3.15.

11. Case Studies

Case study: Netflix Dispatch

Netflix Dispatch is an incident management platform written in Python. It used FastAPI and an async request path, and the repository became a widely cited example of a large asyncio service with real integrations.¹⁵ The repository is archived, so treat it as a historical case study rather than a current dependency.

What to copy: the separation between the request path and background work, so a slow integration does not hold the response open. What to avoid: copying patterns from an archived codebase without checking the current framework version.

Case study: Sentry's asyncio integration

Sentry's Python SDK instruments asyncio tasks, keeps the request context across awaits and reports exceptions from tasks that nobody awaits.¹⁶ The interesting engineering problem is context propagation: a task created in a request must carry the request id and the tracing span even though it runs later.

What to copy: propagate context at task creation time and name tasks after the operation they perform. What to avoid: relying on a global variable for request state, because two concurrent tasks will overwrite it.

Case study: uvloop

uvloop replaces the standard selector loop with a C implementation built on libuv. The project publishes benchmarks that compare throughput and latency against the standard loop.¹⁷ It is a drop-in policy for many applications, and it does not change the async model.

What to copy: measure the loop implementation on your own workload before switching. What to avoid: assuming a faster loop fixes a blocking callback. If the loop is starved, a faster selector does not help.