Python Async Programming: asyncio Patterns for Backend Engineers

snippet-2.py

import asyncio
import httpx

async def fetch_user(client: httpx.AsyncClient, user_id: int) -> dict:
    response = await client.get(f"https://api.internal/users/{user_id}")
    response.raise_for_status()
    return response.json()

async def fetch_permissions(client: httpx.AsyncClient, user_id: int) -> list:
    response = await client.get(f"https://api.internal/permissions/{user_id}")
    response.raise_for_status()
    return response.json()

async def fetch_activity(client: httpx.AsyncClient, user_id: int) -> list:
    response = await client.get(f"https://api.internal/activity/{user_id}")
    response.raise_for_status()
    return response.json()

async def build_user_context(user_id: int) -> dict:
    async with httpx.AsyncClient(timeout=5.0) as client:
        user, permissions, activity = await asyncio.gather(
            fetch_user(client, user_id),
            fetch_permissions(client, user_id),
            fetch_activity(client, user_id),
            return_exceptions=True,  # don't cancel siblings on one failure
        )

    if isinstance(user, Exception):
        raise RuntimeError(f"Failed to fetch user: {user}")

    return {
        "user": user,
        "permissions": permissions if not isinstance(permissions, Exception) else [],
        "activity": activity if not isinstance(activity, Exception) else [],
    }

snippet-3.py

import asyncio
import httpx

async def fetch_with_limit(
    client: httpx.AsyncClient,
    sem: asyncio.Semaphore,
    url: str,
) -> dict:
    async with sem:  # blocks here if semaphore count is exhausted
        response = await client.get(url)
        return response.json()

async def batch_fetch(urls: list[str], max_concurrency: int = 20) -> list[dict]:
    sem = asyncio.Semaphore(max_concurrency)
    async with httpx.AsyncClient() as client:
        tasks = [fetch_with_limit(client, sem, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

snippet-4.py

import asyncio

async def process_event(event: dict) -> None:
    # simulate processing
    await asyncio.sleep(0.1)
    if event.get("poison"):
        raise ValueError(f"Poison event: {event['id']}")

async def handle_batch(events: list[dict]) -> None:
    async with asyncio.TaskGroup() as tg:
        for event in events:
            tg.create_task(process_event(event))
    # if any task raises, all others are cancelled and the exception propagates here

snippet-5.py

import asyncio
import asyncpg

async def get_user(pool: asyncpg.Pool, user_id: int) -> dict | None:
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            "SELECT id, email, created_at FROM users WHERE id = $1",
            user_id,
        )
        return dict(row) if row else None

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@localhost/mydb",
        min_size=5,
        max_size=20,
    )
    try:
        user = await get_user(pool, 42)
        print(user)
    finally:
        await pool.close()

asyncio.run(main())

snippet-6.py

import asyncio
import hashlib
from concurrent.futures import ProcessPoolExecutor

def compute_hash(data: bytes) -> str:
    # CPU-intensive — would block the event loop if called directly
    return hashlib.sha256(data).hexdigest()

async def hash_file(path: str) -> str:
    loop = asyncio.get_running_loop()
    with open(path, "rb") as f:
        data = f.read()
    # offload to process pool to bypass the GIL
    with ProcessPoolExecutor() as pool:
        return await loop.run_in_executor(pool, compute_hash, data)

snippet-7.py

import asyncio
import httpx

async def fetch_with_timeout(url: str, timeout_sec: float = 3.0) -> dict:
    async with httpx.AsyncClient() as client:
        try:
            return await asyncio.wait_for(
                client.get(url),  # this is a coroutine
                timeout=timeout_sec,
            )
        except asyncio.TimeoutError:
            raise RuntimeError(f"Request to {url} timed out after {timeout_sec}s")

snippet.py

import asyncio
import time

async def bad_example():
    # This blocks the event loop for 2 seconds — nothing else runs
    time.sleep(2)

async def good_example():
    # This yields control; other coroutines run during the wait
    await asyncio.sleep(2)

async def main():
    await asyncio.gather(good_example(), good_example(), good_example())

asyncio.run(main())