Python Async Patterns: asyncio, Type Hints, and Production Best Practices

Overview

Modern Python (3.10+) for API integrations, data pipelines, and LLM-powered scripts. Emphasizes type safety and async concurrency.

Type Hints (Always Use)

from typing import Optional, Union, TypeVar, Generic
from collections.abc import Sequence, Callable, AsyncIterator

T = TypeVar("T")

def process_items(
    items: Sequence[str],
    transform: Callable[[str], str],
    limit: Optional[int] = None,
) -> list[str]:
    results = [transform(item) for item in items]
    return results[:limit] if limit else results

# Python 3.10+ union syntax
def get_value(key: str) -> str | None:
    return cache.get(key)

Dataclasses for Structured Data

from dataclasses import dataclass, field
from datetime import datetime

@dataclass
class ApiResponse:
    status: int
    data: dict
    timestamp: datetime = field(default_factory=datetime.utcnow)
    errors: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return 200 <= self.status < 300

Asyncio — Concurrent API Calls

import asyncio
import httpx

async def fetch_one(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url)
    response.raise_for_status()
    return response.json()

async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=30.0) as client:
        tasks = [fetch_one(client, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    # Separate successes from failures
    data, errors = [], []
    for r in results:
        if isinstance(r, Exception):
            errors.append(str(r))
        else:
            data.append(r)

    if errors:
        print(f"Errors: {errors}")
    return data

# Run from sync context
results = asyncio.run(fetch_all(["https://api.example.com/a", "https://api.example.com/b"]))

Context Managers (Resource Safety)

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_session(api_key: str):
    client = httpx.AsyncClient(
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30.0,
    )
    try:
        yield client
    finally:
        await client.aclose()

# Usage
async with managed_session(api_key) as client:
    data = await client.get("/endpoint")

Error Handling Pattern

from enum import Enum

class ErrorCode(Enum):
    NOT_FOUND = "not_found"
    RATE_LIMITED = "rate_limited"
    SERVER_ERROR = "server_error"

class ApiError(Exception):
    def __init__(self, code: ErrorCode, message: str, retry_after: int | None = None):
        super().__init__(message)
        self.code = code
        self.retry_after = retry_after

async def call_api(endpoint: str) -> dict:
    async with httpx.AsyncClient() as client:
        try:
            resp = await client.get(endpoint)
        except httpx.TimeoutException:
            raise ApiError(ErrorCode.SERVER_ERROR, "Request timed out")

        if resp.status_code == 404:
            raise ApiError(ErrorCode.NOT_FOUND, f"Not found: {endpoint}")
        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", "60"))
            raise ApiError(ErrorCode.RATE_LIMITED, "Rate limited", retry_after=retry_after)

        resp.raise_for_status()
        return resp.json()

Retry with Exponential Backoff

import asyncio
import random

async def with_retry(
    coro_fn: Callable,
    max_attempts: int = 3,
    base_delay: float = 1.0,
) -> T:
    for attempt in range(max_attempts):
        try:
            return await coro_fn()
        except ApiError as e:
            if e.code == ErrorCode.RATE_LIMITED:
                delay = e.retry_after or (base_delay * 2 ** attempt + random.random())
                await asyncio.sleep(delay)
            elif attempt == max_attempts - 1:
                raise
    raise RuntimeError("Unreachable")

Environment Variables (Safe Loading)

import os
from dotenv import load_dotenv

load_dotenv()  # Loads .env file if present

def require_env(key: str) -> str:
    value = os.getenv(key)
    if not value:
        raise EnvironmentError(f"Required env var not set: {key}")
    return value

OPENAI_API_KEY = require_env("OPENAI_API_KEY")
SUPABASE_URL = require_env("SUPABASE_URL")

Production Checklist

• All public functions have type hints (mypy or pyright clean)
• Async I/O for all network calls (no blocking requests in async code)
• Context managers for all resource handles (files, HTTP clients, DB connections)
• Retry logic on rate-limited endpoints
• Secrets via os.getenv() — never hardcoded
• Error types are specific, not bare Exception