Async Python Without the Hand-Waving: A Practical Deep Dive into asyncio

Most Python developers meet asyncio the same way: they copy an example with async def and await, it works, and then something breaks in a way that makes no sense. A request hangs. A task silently disappears. Adding async everywhere somehow makes the program slower.

The confusion almost always comes from one missing mental model. This article builds that model, then walks through the pieces you actually use day to day — tasks, gather, TaskGroup, timeouts, queues, and cancellation — with runnable code at every step.

The one idea that makes asyncio click

asyncio gives you concurrency, not parallelism. Everything runs on a single thread, in a single event loop. There is exactly one moment where the loop can switch from one piece of work to another: when your code hits an await on something that isn't ready yet.

That's the whole rule. await means "I'm pausing here — loop, go run something else." If a chunk of code never awaits, nothing else on the loop can run until it finishes.

This is why asyncio shines for I/O-bound work (network calls, database queries, file transfers) where your program spends most of its life waiting, and does nothing at all for CPU-bound work like parsing a huge file or crunching numbers.

Coroutines don't run until you await them

Calling an async def function doesn't execute it. It builds a coroutine object and hands it to you, inert:

import asyncio

async def greet(name):
    await asyncio.sleep(0.1)
    return f"Hello, {name}"

coro = greet("Ada")     # nothing has happened yet
print(coro)             # <coroutine object greet at 0x...>

print(asyncio.run(greet("Ada")))   # Hello, Ada

asyncio.run() is your entry point: it starts an event loop, runs the coroutine to completion, then shuts the loop down cleanly. Call it once, at the top of your program. Don't call it inside async code.

Running things concurrently with gather

Awaiting coroutines one after another is just sequential code with extra syntax. Concurrency starts when you hand several awaitables to the loop at once:

import asyncio, time

async def fetch(name, delay):
    await asyncio.sleep(delay)
    return f"{name} done"

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

asyncio.run(main())
# ['a done', 'b done', 'c done'] 0.31s

Three 0.3-second waits finish in 0.3 seconds total. gather returns results in the order you passed the arguments, not the order they completed.

By default, the first exception propagates immediately. Pass return_exceptions=True when you'd rather collect failures alongside successes:

async def boom():
    raise ValueError("nope")

results = await asyncio.gather(fetch("a", 0.1), boom(), return_exceptions=True)
# ['a done', ValueError('nope')]

Tasks: work that runs in the background

asyncio.create_task() schedules a coroutine on the loop right away and gives you a handle. This is how you start something now and collect it later:

async def main():
    task = asyncio.create_task(fetch("background", 0.5), name="bg")
    # ... do other awaits here; the task makes progress meanwhile ...
    await asyncio.sleep(0.1)
    print(task.done())        # False
    print(await task)         # background done

Pitfall: the disappearing task

The event loop only keeps a weak reference to tasks. If you create one and drop your reference to it, it can be garbage collected mid-flight — the classic "my background job randomly doesn't run" bug. Keep a reference:

_background = set()

def spawn(coro):
    task = asyncio.create_task(coro)
    _background.add(task)
    task.add_done_callback(_background.discard)
    return task

TaskGroup: the modern default (Python 3.11+)

Since Python 3.11, asyncio.TaskGroup is the better choice for running a batch of tasks. It guarantees that no task outlives the block, and if one fails, the rest are cancelled automatically:

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("a", 0.2))
        t2 = tg.create_task(fetch("b", 0.4))
    # both are guaranteed finished here
    print(t1.result(), t2.result())

If several tasks fail, you get an ExceptionGroup, which you handle with except*:

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(boom())
        tg.create_task(fetch("a", 0.1))
except* ValueError as eg:
    print("failures:", eg.exceptions)

Use TaskGroup when tasks belong together and a failure should abort the batch. Use gather(..., return_exceptions=True) when they're independent and you want every result regardless.

Timeouts

asyncio.wait_for wraps a single awaitable and raises TimeoutError if it overruns:

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

try:
    await asyncio.wait_for(slow(), timeout=0.2)
except asyncio.TimeoutError:
    print("timed out")

On Python 3.11+, asyncio.timeout() is more flexible — it applies a deadline to a whole block, however many awaits it contains:

async with asyncio.timeout(2.0):
    data = await fetch_user()
    profile = await fetch_profile(data)

Reacting to results as they arrive

as_completed yields awaitables in completion order, so you can start processing the fast responses without waiting for the slow ones:

async def job(n):
    await asyncio.sleep(0.3 - n * 0.1)
    return n

for coro in asyncio.as_completed([job(0), job(1), job(2)]):
    print("got", await coro)
# got 2
# got 1
# got 0

Throttling with a Semaphore

Firing 5,000 concurrent requests at an API is a good way to get rate-limited. A semaphore caps how many coroutines are inside a section at once:

async def worker(sem, i):
    async with sem:
        await asyncio.sleep(0.1)   # stands in for an HTTP call
        return i * 2

async def main():
    sem = asyncio.Semaphore(3)     # at most 3 in flight
    results = await asyncio.gather(*(worker(sem, i) for i in range(9)))
    print(results)

The pitfall that ruins most async code: blocking calls

This is the big one. A single synchronous call freezes the entire event loop — every other task stops dead:

import time

async def bad():
    time.sleep(2)        # blocks EVERYTHING for 2 seconds

The usual offenders are time.sleep, requests, synchronous database drivers, plain open() on slow storage, and heavy CPU loops. The fix is to push them off the loop with asyncio.to_thread:

def blocking_call():
    time.sleep(0.2)
    return "finished"

async def main():
    result = await asyncio.to_thread(blocking_call)
    print(result)

For CPU-bound work, threads won't help much because of the GIL — reach for loop.run_in_executor() with a ProcessPoolExecutor instead.

Producer/consumer with asyncio.Queue

asyncio.Queue is the clean way to decouple tasks that generate work from tasks that process it:

async def producer(q):
    for i in range(5):
        await q.put(i)
    await q.put(None)          # sentinel

async def consumer(q):
    out = []
    while True:
        item = await q.get()
        if item is None:
            break
        out.append(item)
    return out

async def main():
    q = asyncio.Queue()
    _, got = await asyncio.gather(producer(q), consumer(q))
    print(got)                 # [0, 1, 2, 3, 4]

Cancellation is an exception, not a kill switch

Cancelling a task raises CancelledError inside it at its next await point. That means cleanup code still runs — and that you should let the exception propagate rather than swallowing it:

async def main():
    task = asyncio.create_task(slow())
    await asyncio.sleep(0.05)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("cancelled cleanly")

Put teardown in a finally block so it survives cancellation:

async def worker():
    conn = await open_connection()
    try:
        await do_work(conn)
    finally:
        await conn.close()     # runs even if cancelled

Async generators and context managers

Two smaller pieces of the async protocol worth knowing. Async generators let you stream results lazily with async for:

async def ticker(n):
    for i in range(n):
        await asyncio.sleep(0.05)
        yield i

values = [x async for x in ticker(3)]   # [0, 1, 2]

And __aenter__ / __aexit__ give you async with for resources whose setup and teardown are themselves awaitable:

class Conn:
    async def __aenter__(self):
        await asyncio.sleep(0.01)
        print("opened")
        return self

    async def __aexit__(self, *exc):
        await asyncio.sleep(0.01)
        print("closed")
        return False

async def main():
    async with Conn() as conn:
        ...

A quick checklist

• One asyncio.run() per program, at the top.
• Prefer TaskGroup (3.11+) over bare create_task for batches.
• Keep references to fire-and-forget tasks.
• Never call blocking code on the loop — wrap it in to_thread.
• Bound concurrency with a Semaphore when talking to external services.
• Put cleanup in finally, and let CancelledError propagate.
• Set a timeout on anything that touches the network.

Wrap-up and next steps

asyncio isn't magic and it isn't a speed-up button. It's a way to keep one thread busy while it would otherwise sit waiting on I/O — and once you internalise that await is the only place the loop can switch, the rest of the API stops feeling arbitrary.

From here, the natural next steps are picking up an async-native library and building something real: httpx or aiohttp for HTTP clients, asyncpg for Postgres, and FastAPI or Starlette for servers. Also try running your code with asyncio.run(main(), debug=True) — debug mode warns you about coroutines that were never awaited and callbacks that blocked the loop for too long, which catches a surprising share of async bugs before they reach production.