Async/Await — How the Event Loop Works

asyncio gives you concurrency without threads. Critical for I/O-bound workloads; useless for CPU-bound ones.

The mental model

Async functions (coroutines) suspend at await points; control returns to the event loop, which runs other ready coroutines.

import asyncio

async def fetch(url):
    print(f"Fetching {url}")
    await asyncio.sleep(1)   # simulate I/O
    print(f"Done {url}")
    return f"data from {url}"

async def main():
    results = await asyncio.gather(
        fetch("https://a.com"),
        fetch("https://b.com"),
        fetch("https://c.com"),
    )
    print(results)

asyncio.run(main())

Output: all three "Fetching..." print, then ~1 second later all three "Done...". Total time: ~1 second, not 3.

Single thread, single event loop, concurrent I/O.

When asyncio helps

I/O-bound workloads — when your code spends most of its time waiting:

• HTTP requests (lots of network).
• Database queries (waiting on the DB).
• File I/O (especially many files).
• WebSocket / SSE servers.
• Pub-sub consumers.

Speedup ratio: 10-100x over synchronous for I/O-heavy code.

When asyncio DOESN'T help

CPU-bound workloads — code that's actually computing, not waiting:

• Number crunching.
• Image/video processing (in pure Python).
• ML inference.
• Sorting large arrays.

asyncio doesn't help because there's only one thread; the CPU is the bottleneck.

For CPU-bound: use multiprocessing (next lesson).

The key rule: don't block the event loop

If you call a blocking function inside async code, you freeze the whole event loop. Other coroutines stall.

Bad:

async def handler():
    time.sleep(1)        # blocks event loop for 1 sec
    result = requests.get("https://api.com")   # blocks

Good:

async def handler():
    await asyncio.sleep(1)            # non-blocking
    async with httpx.AsyncClient() as client:
        result = await client.get("https://api.com")   # non-blocking

Use async-aware libraries: httpx (not requests), asyncpg (not psycopg2), aiofiles (not open).

For CPU-bound work mid-async: offload to a thread pool:

result = await asyncio.to_thread(blocking_function, args)

Common patterns

Concurrent execution with gather

results = await asyncio.gather(
    fetch("a"),
    fetch("b"),
    fetch("c"),
)

All three run concurrently; result list in input order.

Bounded concurrency

async def fetch_with_semaphore(sem, url):
    async with sem:
        return await fetch(url)

sem = asyncio.Semaphore(10)   # max 10 concurrent
tasks = [fetch_with_semaphore(sem, url) for url in urls]
results = await asyncio.gather(*tasks)

Prevents overwhelming the target with 1000 concurrent requests.

Timeouts

try:
    result = await asyncio.wait_for(slow_operation(), timeout=5)
except asyncio.TimeoutError:
    # handle timeout
    pass

Critical for any external call.

TaskGroup (Python 3.11+)

async with asyncio.TaskGroup() as tg:
    tg.create_task(work1())
    tg.create_task(work2())
    # All tasks run concurrently; exceptions propagate cleanly

Modern replacement for raw gather; better error handling.

Async libraries you'll use

For	Use
HTTP client	httpx, aiohttp
HTTP server	FastAPI, Starlette, aiohttp
Postgres	asyncpg, psycopg (3.0+)
Redis	redis.asyncio
Kafka	aiokafka
File I/O	aiofiles
Threading bridge	asyncio.to_thread

Sync libraries inside async = bad. Pick the async variant.

Async + sync interop

You can call sync from async:

result = await asyncio.to_thread(blocking_fn, arg)

You can't directly call async from sync (without setting up an event loop):

result = asyncio.run(async_fn())   # OK
result = async_fn()                 # NO, returns coroutine

In a sync codebase wanting to call async lib functions: asyncio.run() at the entry point.

Common asyncio mistakes

1. Calling sync I/O inside async. Blocks everything. Use async libraries.
2. Forgetting await. Function call returns a coroutine object, doesn't execute. Common bug.
3. Using asyncio for CPU-bound work. No speedup; complexity for nothing.
4. No timeouts on external calls. Hangs forever on slow services.
5. No bounded concurrency. Spawning 10,000 tasks at once; OOM or overwhelmed targets.
6. Mixing sync and async carelessly. time.sleep() in async = 🚨.

Forgetting await — the silent killer

async def main():
    fetch_url("https://a.com")    # returns coroutine; doesn't run!
    print("done")

# vs
async def main():
    await fetch_url("https://a.com")    # actually runs
    print("done")

Pyright/mypy catches some of these (unused coroutine warning); rely on the type checker.

When asyncio is overkill

For simple sequential I/O (one or two API calls), use sync. Async adds complexity:

• Async-flavored libraries everywhere.
• Harder debugging (stack traces are messier).
• More mental overhead.

Threshold: 5+ concurrent I/O operations or sustained high-volume I/O → use asyncio.

asyncio vs trio / anyio

asyncio is stdlib. Alternatives:

• trio: cleaner API, structured concurrency built-in.
• anyio: abstraction over asyncio + trio.

For most production use: stdlib asyncio. trio if you appreciate cleaner semantics.

Takeaway

asyncio = single-threaded concurrency for I/O-bound work. Coroutines yield at await. Use async libraries (httpx, asyncpg); never sync I/O inside async. Patterns: gather, semaphore for bounded concurrency, timeouts, TaskGroup. Forgot-to-await is a classic bug. For CPU-bound: multiprocessing, not asyncio.