Async Programming in Python: asyncio Explained

Most Python programs spend their time waiting: for an API to answer, a database to return rows, a file to arrive. asyncio lets one thread wait on many of those things at once. It does not make your code faster at computing anything. It makes waiting overlap. This guide shows what that means with timings you can run, then covers the parts that trip people up: tasks, timeouts, limiting concurrency, and the blocking call that quietly undoes everything.

Three waits, one second

A coroutine is a function defined with async def. Calling it does not run it; it returns a coroutine object that runs when something awaits it. asyncio.run() starts the event loop and runs your top-level coroutine (Python docs: coroutines and tasks).

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)          # stands in for a network call
    return f"{name} done"

async def one_by_one():
    start = time.perf_counter()
    for name in ("a", "b", "c"):
        await fetch(name, 1)
    print(f"sequential: {time.perf_counter() - start:.1f}s")

async def together():
    start = time.perf_counter()
    results = await asyncio.gather(fetch("a", 1), fetch("b", 1), fetch("c", 1))
    print(results, f"{time.perf_counter() - start:.1f}s")

asyncio.run(one_by_one())
asyncio.run(together())

sequential: 3.0s
['a done', 'b done', 'c done'] 1.0s

Awaiting one coroutine after another is still sequential. gather schedules all three, so their waits overlap. At each await, a coroutine hands control back to the event loop, which runs whatever else is ready. That is the whole trick: cooperative switching at await points, on one thread.

Coroutines, tasks and TaskGroup

A task is a coroutine the event loop has scheduled to run on its own. asyncio.create_task() starts it immediately in the background and returns a handle you can await later. gather wraps its arguments in tasks for you.

Since Python 3.11, asyncio.TaskGroup is the structured way to run several tasks. When the async with block ends, every task has finished. If one task fails, the group cancels the others and raises the errors together as an ExceptionGroup:

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

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(might_fail(n)) for n in (1, 2, 3)]
    except* ValueError as eg:
        print("errors:", [str(e) for e in eg.exceptions])
        print("task 3 cancelled:", tasks[2].cancelled())

asyncio.run(main())

errors: ['task 2 failed']
task 3 cancelled: True

With gather, by contrast, the first exception propagates to the awaiting code, but the other tasks keep running unless you cancel them. Prefer TaskGroup in new code on Python 3.11 or later. One more rule from the documentation: keep a reference to tasks you create with create_task, because the event loop keeps only a weak reference and an unreferenced task can disappear mid-run.

The blocking call that makes async sequential

The event loop can only switch at an await. Anything that blocks without awaiting, such as time.sleep(), requests.get() or a heavy calculation, holds the only thread, and every other coroutine waits behind it. The Python documentation warns against calling blocking code directly from a coroutine for exactly this reason (Python docs: developing with asyncio).

async def blocking_fetch(name: str) -> str:
    time.sleep(1)                       # blocks the whole event loop
    return f"{name} done"

async def offloaded_fetch(name: str) -> str:
    await asyncio.to_thread(time.sleep, 1)   # runs in a worker thread
    return f"{name} done"

async def compare():
    for fn in (blocking_fetch, offloaded_fetch):
        start = time.perf_counter()
        await asyncio.gather(*(fn(n) for n in "abc"))
        print(f"{fn.__name__}: {time.perf_counter() - start:.1f}s")

asyncio.run(compare())

blocking_fetch: 3.0s
offloaded_fetch: 1.0s

Nothing fails and no warning is printed; the program is simply as slow as the synchronous version. The fixes, in order of preference:

• Use a library built for asyncio: an async HTTP client instead of requests, an async database driver instead of a synchronous one.
• Wrap an unavoidable blocking call in asyncio.to_thread() (Python 3.9+), as above.
• Send CPU-heavy work to a process pool. asyncio does not help with computation at all.

Running with asyncio.run(main(), debug=True) makes the event loop log callbacks that take longer than 100 milliseconds, which is how you find the blocking call in a larger program.

Timeouts and a concurrency limit

Two controls matter as soon as the waits are real network calls. The first is a timeout, so one stuck request cannot hang the program. On Python 3.11+, asyncio.timeout() wraps a block; asyncio.wait_for() wraps a single awaitable on older versions:

async def slow():
    await asyncio.sleep(5)

async def with_timeout():
    try:
        async with asyncio.timeout(0.5):
            await slow()
    except TimeoutError:
        print("gave up after 0.5s")

asyncio.run(with_timeout())

gave up after 0.5s

The second is a concurrency limit. gather over 10,000 URLs opens 10,000 requests at once, which can exhaust sockets or trip the server's rate limit. An asyncio.Semaphore lets only a fixed number of coroutines into a block at a time (Python docs: synchronization primitives):

async def limited_fetch(sem: asyncio.Semaphore, i: int, active: list) -> int:
    async with sem:                      # at most 3 inside at once
        active.append(i)
        peak = len(active)
        await asyncio.sleep(0.1)
        active.remove(i)
        return peak

async def crawl():
    sem = asyncio.Semaphore(3)
    active: list[int] = []
    peaks = await asyncio.gather(*(limited_fetch(sem, i, active) for i in range(10)))
    print("most at once:", max(peaks))

asyncio.run(crawl())

most at once: 3

When asyncio is the right tool

asyncio shines when the work is many independent I/O waits. The official documentation describes it as a good fit for I/O-bound and high-level structured network code (Python docs: asyncio). For how threads and processes compare, including the global interpreter lock, see multithreading vs multiprocessing in Python.

Common mistakes

• Calling a coroutine without awaiting it. fetch("a", 1) on its own does nothing, and Python warns that the coroutine was never awaited.
• Using asyncio.run() inside code that is already async, such as a Jupyter cell or a FastAPI endpoint. There, just await the coroutine.
• Mixing in synchronous libraries without to_thread: the blocking call above.
• Unbounded fan-out. Always put a semaphore, or a fixed pool of worker tasks reading from an asyncio.Queue, in front of large batches.

Next steps

asyncio sits in the concurrency module of the Python Programming course, alongside threading and multiprocessing. Async endpoints are also where API frameworks use it; the FastAPI guide shows one in practice.

Sources

• Coroutines and Tasks — Python documentation
• Developing with asyncio — Python documentation
• Synchronization Primitives — Python documentation
• asyncio — Asynchronous I/O — Python documentation