Python asyncio

Single-threaded concurrency via cooperative multitasking — one thread, one event loop, many coroutines. Best for I/O-bound work (HTTP, DB, file reads). For when to reach for asyncio over threading/multiprocessing, see [[Python Concurrency]]'s decision framework — this page is the mechanics once you've decided asyncio is the right tool.

Core Concepts

import asyncio

async def fetch(url: str) -> str:
    await asyncio.sleep(1)   # simulate I/O — yields control back to the event loop
    return f"data from {url}"

async def main():
    result = await fetch("https://api.example.com")
    print(result)

asyncio.run(main())   # entry point — creates the event loop, runs until done

• async def → a coroutine function; calling it returns a coroutine object, doesn't run the body yet.
• await → yield control until the awaited thing completes; only valid inside async def.
• asyncio.run() → the top-level entry point — call it once, at the top of the program, never from inside already-running async code.

Concurrent I/O with gather

Sequentially await-ing each call in a loop is still sequential — gather is what makes multiple I/O-bound calls actually overlap.

import asyncio, aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def main():
    urls = [f"https://api.example.com/users/{i}" for i in range(1, 4)]
    async with aiohttp.ClientSession() as session:
        # Sequential: 3 x 200ms = 600ms
        # results = [await fetch(session, url) for url in urls]

        # Concurrent: max(200ms each) ~ 200ms total
        results = await asyncio.gather(*[fetch(session, url) for url in urls])
    return results

gather vs wait vs TaskGroup

# gather — all results, or the first exception cancels the rest
results = await asyncio.gather(coro1(), coro2(), return_exceptions=True)

# wait — fine-grained control over completion condition
done, pending = await asyncio.wait(
    {asyncio.create_task(c) for c in coros},
    timeout=5.0, return_when=asyncio.FIRST_COMPLETED,
)

# TaskGroup (Python 3.11+) — structured concurrency, cancels siblings on any failure
async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(coro1())
    task2 = tg.create_task(coro2())
# guaranteed: every task is done (or cancelled) by the time the block exits

Semaphore — Capping Concurrency

async def fetch_limited(session, url, sem):
    async with sem:                    # only N concurrent requests at once
        return await fetch(session, url)

async def main():
    sem = asyncio.Semaphore(10)        # cap at 10 concurrent
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[fetch_limited(session, url, sem) for url in urls])

Uncapped concurrency against thousands of URLs can exhaust file descriptors or trip a downstream rate limiter — a semaphore is the standard fix, the async-native equivalent of a connection pool size limit.

Async Context Managers & Iterators

class AsyncDB:
    async def __aenter__(self):
        self.conn = await create_connection()
        return self.conn
    async def __aexit__(self, *args):
        await self.conn.close()

async with AsyncDB() as conn:
    await conn.execute("SELECT ...")

class AsyncStream:
    async def __aiter__(self):
        return self
    async def __anext__(self):
        data = await self.read_chunk()
        if not data:
            raise StopAsyncIteration
        return data

async for chunk in AsyncStream():
    process(chunk)

asyncio.Queue — Producer/Consumer

async def producer(queue: asyncio.Queue):
    for i in range(10):
        await queue.put(i)
        await asyncio.sleep(0.1)
    await queue.put(None)              # sentinel — signals "no more items"

async def consumer(queue: asyncio.Queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        process(item)
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)   # bounded queue — backpressure: producer blocks when full
    await asyncio.gather(producer(queue), consumer(queue))

A bounded maxsize is what gives backpressure — an unbounded queue would let a fast producer pile up unlimited memory ahead of a slow consumer; capping it forces the producer to wait, which is usually the correct behavior.

The Event Loop, Concretely

Event Loop
├── Task 1: fetch(url1) → awaits network → suspended
├── Task 2: fetch(url2) → awaits network → suspended
├── Task 3: fetch(url3) → awaits network → suspended
└── When I/O ready → resume that task → run until its next await

Only one coroutine executes Python code at any instant — this is cooperative, not preemptive, concurrency. CPU-bound code inside a coroutine blocks the entire event loop — nothing else runs until it returns, since there's no await point for the scheduler to interrupt it at.

Running Blocking/CPU-Bound Code From Async Code

import asyncio
from concurrent.futures import ProcessPoolExecutor

async def fetch_sync_data():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_db_call, arg1)   # thread pool — I/O-bound
    return result

async def compute():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, cpu_heavy_fn, data)  # process pool — CPU-bound
    return result

Full GIL/threading/multiprocessing reasoning behind this split is in [[Python Concurrency]].

Common Mistakes

Production Example — FastAPI

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    user = await db.get(User, user_id)    # async SQLAlchemy
    return user

See [[FastAPI]] for the full request-handling picture this fits into.

Interview Drills

1. Why doesn't the GIL matter for asyncio's performance model? — asyncio is single-threaded by design — the GIL is a lock about multiple threads contending for the interpreter, and with only one thread there's no contention to resolve; asyncio's concurrency comes entirely from cooperative yielding at await points, a different mechanism than thread scheduling.
2. A route handler does results = [await fetch(u) for u in urls] for 5 URLs. Why is this slower than expected, and what's the fix? — Each await in the list comprehension runs to completion before the next iteration starts — this is sequential, not concurrent, despite using async/await. Fix: await asyncio.gather(*[fetch(u) for u in urls]), which starts all coroutines and waits for them together.
3. Why does a bounded asyncio.Queue(maxsize=5) matter for a producer/consumer pipeline? — It provides backpressure — if the producer is faster than the consumer, queue.put() blocks once the queue is full, pausing the producer rather than letting memory grow unbounded; an unbounded queue would hide a real capacity mismatch instead of surfacing it as controlled backpressure.

Cross-links

[[Python Concurrency]] · [[FastAPI]] · [[Pydantic]] · [[Go Concurrency]]