Python Async/Await: From Basics to Production Patterns

Why Async Python?

Python's asyncio allows you to write concurrent code that handles thousands of I/O operations simultaneously within a single thread. This is ideal for web servers, API clients, database queries, and file operations where most time is spent waiting for external responses rather than computing.

The key mental model: async def defines a coroutine (a function that can pause and resume), and await is the pause point where the event loop can switch to other tasks while waiting for I/O to complete.

Sync vs Async Comparison

import requests
import time

urls = [f"https://api.example.com/{i}"
        for i in range(100)]

start = time.time()
results = []
for url in urls:
    r = requests.get(url)
    results.append(r.json())
# Total: ~100 seconds (1s per request)

import aiohttp, asyncio

async def fetch_all():
    async with aiohttp.ClientSession() as s:
        tasks = [s.get(f"https://api.example.com/{i}")
                 for i in range(100)]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]

results = asyncio.run(fetch_all())
# Total: ~1.5 seconds (parallel)

Core Concepts

1. Coroutines

A coroutine is defined with async def. Calling it does not execute it—it returns a coroutine object that must be awaited or scheduled:

async def greet(name):
    await asyncio.sleep(1)  # Simulate I/O
    return f"Hello, {name}!"

# Wrong: greet("Alice") returns a coroutine object, not a string
# Right: await greet("Alice") or asyncio.run(greet("Alice"))

2. Tasks

Tasks wrap coroutines and schedule them to run concurrently on the event loop:

async def main():
    task1 = asyncio.create_task(fetch_data("userA"))
    task2 = asyncio.create_task(fetch_data("userB"))
    # Both run concurrently
    result1 = await task1
    result2 = await task2

3. gather vs TaskGroup

asyncio.gather() runs multiple coroutines concurrently. In Python 3.11+, TaskGroup provides structured concurrency with better error handling:

# Python 3.11+ TaskGroup (preferred)
async with asyncio.TaskGroup() as tg:
    task1 = tg.create_task(fetch("url1"))
    task2 = tg.create_task(fetch("url2"))
# If any task fails, all others are cancelled

Rate Limiting with Semaphores

When making thousands of concurrent requests, you'll overwhelm APIs or run out of file descriptors. Semaphores limit concurrency:

sem = asyncio.Semaphore(50)  # Max 50 concurrent operations

async def limited_fetch(session, url):
    async with sem:
        async with session.get(url) as response:
            return await response.json()

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

Common Pitfalls

Production Patterns

• Retry with backoff: Use tenacity library with async support for automatic retries on network failures.
• Graceful shutdown: Handle SIGTERM to cancel running tasks and close connections before exit.
• Connection pooling: Create a single aiohttp.ClientSession and reuse it across all requests. Don't create one per request.
• Timeout protection: Always use asyncio.wait_for(coro, timeout=30) to prevent coroutines from hanging indefinitely.
• Structured logging: Use structlog with async-safe handlers. Standard logging is thread-safe but not always async-friendly.

For documenting async architectures in technical articles, SciDraw generates event loop diagrams, coroutine flow charts, and concurrency pattern visualizations. PatentFig is ideal for patent-quality technical diagrams.