Python asyncio Recipes

I like theory. I also like shipping code that works. With asyncio, the fastest way to ship is to grab a few solid patterns and use them.

Here are the patterns I actually use. Start by copying something that runs. If you want the deeper event loop explanation, go read my theoretical async post after you've got something working: Asyncio: the theory (after you ship).

1) Fetch multiple URLs concurrently

Use this for I/O-bound work where you don't want to wait on each request one at a time.

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as r:
        r.raise_for_status()
        return await r.text()

async def main(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, u) for u in urls]
        return await asyncio.gather(*tasks)

## asyncio.run(main([...]))

Common mistake: creating a new ClientSession per request. Create one session and reuse it.

2) Run a task with a timeout

Use this when an API sometimes hangs and you'd rather get an error than wait forever.

import asyncio

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

async def main():
    try:
        return await asyncio.wait_for(do_work(), timeout=1.0)
    except asyncio.TimeoutError:
        return "timed out"

Common mistake: forgetting that a task you created elsewhere may keep running. wait_for() cancels the awaited coroutine, but if you spun up a separate task, you still need to manage its lifecycle.

3) Process a queue with a fixed worker pool

Use this for "N things to do" where each thing is mostly I/O, and you want controlled concurrency.

import asyncio

async def worker(name, q):
    while True:
        item = await q.get()
        try:
            await asyncio.sleep(0.1)  # pretend I/O
            print(name, "processed", item)
        finally:
            q.task_done()

async def main(items, workers=5):
    q = asyncio.Queue()
    for it in items:
        q.put_nowait(it)

    tasks = [asyncio.create_task(worker(f"w{i}", q)) for i in range(workers)]
    await q.join()

    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)

Common mistake: forgetting task_done() and then wondering why join() never returns.

4) Debounce rapid calls

Use this when an event fires 30 times (file watcher, keystrokes, websocket messages) and you only want the last one.

import asyncio

class Debouncer:
    def __init__(self, delay):
        self.delay = delay
        self._task = None

    def call(self, coro_func, *args, **kwargs):
        if self._task:
            self._task.cancel()

        async def runner():
            await asyncio.sleep(self.delay)
            await coro_func(*args, **kwargs)

        self._task = asyncio.create_task(runner())

Common mistake: putting sleep() in the hot path so you stall unrelated work. Debounce should schedule work, not pause everything.

5) Run blocking code without blocking the loop

Use this when you have a sync library (or some "quick" CPU work) and you still need the event loop to keep serving other tasks.

import asyncio
import time

def blocking():
    time.sleep(2)
    return 123

async def main():
    result = await asyncio.to_thread(blocking)
    return result

Common mistake: calling blocking code directly inside async def. If you see time.sleep() in async code, you've found the bottleneck.

6) Graceful shutdown (signals + cleanup)

Use this for servers and workers. You want Ctrl+C to mean "stop accepting new work and exit cleanly."

import asyncio
import signal

async def run(stop_event):
    while not stop_event.is_set():
        await asyncio.sleep(0.2)

async def main():
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()

    for s in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(s, stop_event.set)

    await run(stop_event)

## asyncio.run(main())

Common mistake: catching KeyboardInterrupt and assuming shutdown is handled. Your tasks may still be running unless you coordinate cancellation and cleanup.

7) Mix async code with sync libraries that don't support it

Use this when you're stuck with a sync SDK, a legacy DB client, or a "simple" function that takes 800ms.

import asyncio
from functools import partial

def sync_api_call(x, y):
    return x + y

async def main():
    fn = partial(sync_api_call, 2, 3)
    return await asyncio.to_thread(fn)

Common mistake: wrapping a sync function in async def and calling it a day. If the function blocks, it still blocks.

If you're new to this, here's the rule I use: get something running early, then tighten it up. Most asyncio pain comes from two things: accidentally serializing I/O, or blocking the loop. These recipes cover both.

Ship something that works first. Read the theory later if you're curious how it works under the hood. When you're ready to triage and fix what you shipped, you'll have real production feedback to guide your improvements.

-Sethers