asyncio

asyncio is cooperative concurrency: one thread, one event loop, and potentially thousands of I/O operations in flight at once. The key player is the coroutine — a function defined with async def that can pause partway through. When a coroutine waits — on a socket, a timer, a database response — it yields control back to the loop, which immediately runs the next ready coroutine. No GIL contention, no thread overhead, very little memory per task.

The one limitation to keep front of mind: because it is single-threaded, asyncio never uses more than one CPU core. Its superpower is eliminating idle waiting, not parallelising computation. For CPU work you still want processes.

Coroutines versus functions

A normal function runs to completion the moment you call it. A coroutine function is different: calling it returns a coroutine object that has not run yet — it needs something to drive it, and that something is the event loop:

import asyncio

async def greet(name):
    await asyncio.sleep(0.1)        # yield control while "waiting"
    return f"Hello, {name}!"

async def main():
    # Calling a coroutine function returns a coroutine — nothing has run.
    coro = greet("Aarav")
    print(type(coro).__name__)      # it is a 'coroutine' object

    # await actually runs it and produces the value.
    msg = await greet("Priya")
    print(msg)

asyncio.run(main())

coroutine
Hello, Priya!

asyncio.run(main()) is your program’s single doorway into the loop — it creates the loop, runs main to completion, then shuts the loop down, and you call it exactly once per program. The star, though, is await, which does two things at once: it runs the coroutine, and while that coroutine is waiting, it hands control back to the loop so other tasks can run. (The unawaited greet("Aarav") above is, by the way, the very bug we will meet at the end — it created a coroutine that never ran.)

You can only use await inside an async def, and that restriction is the entire point of the async/await split: it makes the places where your code might yield syntactically visible, so you can see at a glance where control can pass to another task.

Concurrency with gather

A bare await is sequential — it waits for each coroutine in turn before starting the next. To run coroutines genuinely at the same time, hand them to asyncio.gather:

import asyncio

async def slow_fetch(url):
    await asyncio.sleep(0.2)        # pretend this is an HTTP call
    return f"OK {url}"

async def main():
    urls = [f"/u/{i}" for i in range(5)]
    # gather schedules all five, then awaits them together.
    results = await asyncio.gather(*[slow_fetch(u) for u in urls])
    print(results)

asyncio.run(main())

['OK /u/0', 'OK /u/1', 'OK /u/2', 'OK /u/3', 'OK /u/4']

Reason about the timing instead of measuring it: awaiting the five fetches one at a time would take about 5 × 0.2 = 1.0s, since each waits its full 0.2s before the next begins. But gather starts all five, and their waits overlap, so the whole batch finishes in roughly 0.2s. The results come back in input order regardless of which finished first, and by default, if one coroutine raises, the others keep running and the exception propagates out of gather.

create_task and TaskGroup

asyncio.create_task(coro) is gather’s lower-level cousin: it schedules a coroutine immediately and hands back a Task, which you await later when you actually need the result. It is the tool for genuine background work — kicking something off while you carry on:

async def main():
    task = asyncio.create_task(slow_fetch("/u/1"))   # starts running now
    # ... do other work meanwhile ...
    result = await task                              # join when you need it

For most concurrent work in Python 3.11+, though, prefer asyncio.TaskGroup — structured concurrency. The async with block guarantees every task finishes (or all are cancelled together) before it exits, and it surfaces failures cleanly as an ExceptionGroup:

import asyncio

async def fetch(url):
    await asyncio.sleep(0.1)
    if "fail" in url:
        raise ConnectionError(url)
    return f"OK {url}"

async def main():
    urls = ["/a", "/b", "/c"]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(u)) for u in urls]
    # The 'async with' block waits for every task before moving on.
    for t in tasks:
        print(t.result())

asyncio.run(main())

OK /a
OK /b
OK /c

TaskGroup makes the scope of concurrent work explicit — when the block ends, the work is done — which is precisely what you want for clean cancellation and for debugging.

A realistic fan-out with httpx

Real async HTTP needs an async client, and httpx.AsyncClient is the standard choice. The pattern is always the same: open one client, fan out, gather:

import asyncio, httpx

async def fetch_json(client, url):
    r = await client.get(url, timeout=10)
    r.raise_for_status()
    return r.json()

async def main():
    urls = [f"https://httpbin.org/anything/{i}" for i in range(20)]
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch_json(client, u) for u in urls])
    print(f"Got {len(results)} responses")

asyncio.run(main())

The crucial detail is one client, many requests. Creating a fresh client per request destroys throughput, because every call then re-does the TLS handshake from scratch; async with httpx.AsyncClient() keeps a connection pool open for the whole block and reuses it.

The “I forgot to await” bug, and blocking the loop

Every async beginner hits this once: calling a coroutine without awaiting it does not run it — it just makes an unused coroutine object.

import asyncio

async def do_work():
    await asyncio.sleep(0.05)
    return "done!"

async def main():
    # WRONG — the coroutine is created but never awaited.
    result = do_work()
    print("bug:    ", result)

    # RIGHT — await runs it and returns the value.
    result = await do_work()
    print("correct:", result)

asyncio.run(main())

bug:     <coroutine object do_work at 0x...>
correct: done!

There is the answer to the prediction: the missing await left result holding the coroutine object itself (the 0x... is a memory address that varies each run), and Python additionally emits a RuntimeWarning: coroutine '...' was never awaited. If you ever see that printed where you expected a value, you dropped an await.

A related trap is blocking the loop. A plain time.sleep(1) inside an async function freezes the entire event loop — every other task is stalled for that whole second — and the same goes for any synchronous I/O like requests.get or a blocking psycopg2 call. The fix is to use the async equivalent (asyncio.sleep, httpx.AsyncClient, asyncpg), or to push a stubbornly-synchronous call onto a worker thread with asyncio.to_thread(fn, *args), which returns an awaitable while the loop keeps spinning.

In one breath

• A coroutine (async def) does not run until awaited; asyncio.run(main()) is the one entry point.
• await runs a coroutine and yields to the loop while it waits — that is how one thread serves many.
• gather (or TaskGroup in 3.11+) runs coroutines concurrently; bare await in a loop is sequential.
• Use one httpx.AsyncClient for many calls; never block the loop with time.sleep/requests — use async or asyncio.to_thread.
• Forgetting await leaves a coroutine object and a never awaited warning.

Practice

What’s next

asyncio shines brightest on high fan-out I/O — and there is no better case study than async patterns for LLM apps, where every call is a slow round-trip and you want 10–100 of them in flight at once.