Python Academy · Lesson

Async I/O Patterns in Production

Apply asyncio to HTTP clients, queues, and real-world pipelines.

Async I/O Patterns in Production is a free Python Academy lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the Python Academy learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Event-Loop Best Practices

Always use asyncio.run() as the single entry point. Avoid mixing sync and async code carelessly and never block the event loop with long CPU work.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Connection Pooling

Reuse connections instead of opening a new one per request. Libraries like asyncpg and aiohttp provide built-in pools.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Producer-Consumer with asyncio.Queue

asyncio.Queue decouples producers and consumers running as concurrent tasks.

import asyncio

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

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

Rate Limiting with Semaphore

Limit concurrent outbound requests to avoid overwhelming external APIs or hitting rate limits.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

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

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

Retry with Exponential Back-off

Wrap retryable operations with back-off logic to handle transient failures gracefully.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Graceful Shutdown

Catch asyncio.CancelledError in long-running tasks and handle OS signals to shut down cleanly.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Streaming Responses

Use async generators to stream large responses without loading them entirely into memory.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Background Tasks in FastAPI

FastAPI's BackgroundTasks queues work to run after the response is sent, keeping endpoints fast.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Heartbeat Pattern

Run a periodic background task alongside the main server loop using create_task.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Profiling Async Code

Use yappi or py-spy for async-aware profiling. Enable asyncio debug mode for slow-callback warnings.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Avoiding Common Pitfalls

Key rules: never call blocking code on the event loop; always await coroutines; cancel tasks on shutdown; use connection pools; limit concurrency with semaphores.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Quick Check

Which asyncio primitive decouples producers and consumers running as concurrent tasks?

Recap

Production async patterns: use connection pools, control concurrency with Semaphore, implement retry/back-off, stream large data, shut down gracefully on signals, and run periodic tasks with create_task.

Frequently asked questions

Is the “Async I/O Patterns in Production” lesson free?

Yes — the full text of “Async I/O Patterns in Production” is free to read here on the web, and the Python Academy course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the Python Academy course, upgrade to CoddyKit PRO.

What will I learn in “Async I/O Patterns in Production”?

Apply asyncio to HTTP clients, queues, and real-world pipelines. You practise Python Academy with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start Python Academy?

No prior experience is required. Python Academy on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Async I/O Patterns in Production” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this Python Academy lesson?

Yes. Every Python Academy lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

← Back to Python Academy