{"slug":"python-async-patterns","title":"Python Async Patterns: asyncio, Type Hints, and Production Best Practices","tags":["python","asyncio","type-hints","performance","api"],"agent_summary":"Production Python patterns: asyncio concurrency, type hints, dataclasses, context managers, and error handling for API integrations and data pipelines.","trigger_phrases":["Python asyncio","async await Python","Python type hints","Python dataclass","Python context manager","Python concurrency"],"runnable":false,"markdown":"\n## Overview\n\nModern Python (3.10+) for API integrations, data pipelines, and LLM-powered scripts. Emphasizes type safety and async concurrency.\n\n## Type Hints (Always Use)\n\n```python\nfrom typing import Optional, Union, TypeVar, Generic\nfrom collections.abc import Sequence, Callable, AsyncIterator\n\nT = TypeVar(\"T\")\n\ndef process_items(\n items: Sequence[str],\n transform: Callable[[str], str],\n limit: Optional[int] = None,\n) -> list[str]:\n results = [transform(item) for item in items]\n return results[:limit] if limit else results\n\n# Python 3.10+ union syntax\ndef get_value(key: str) -> str | None:\n return cache.get(key)\n```\n\n## Dataclasses for Structured Data\n\n```python\nfrom dataclasses import dataclass, field\nfrom datetime import datetime\n\n@dataclass\nclass ApiResponse:\n status: int\n data: dict\n timestamp: datetime = field(default_factory=datetime.utcnow)\n errors: list[str] = field(default_factory=list)\n\n @property\n def ok(self) -> bool:\n return 200 <= self.status < 300\n```\n\n## Asyncio — Concurrent API Calls\n\n```python\nimport asyncio\nimport httpx\n\nasync def fetch_one(client: httpx.AsyncClient, url: str) -> dict:\n response = await client.get(url)\n response.raise_for_status()\n return response.json()\n\nasync def fetch_all(urls: list[str]) -> list[dict]:\n async with httpx.AsyncClient(timeout=30.0) as client:\n tasks = [fetch_one(client, url) for url in urls]\n results = await asyncio.gather(*tasks, return_exceptions=True)\n\n # Separate successes from failures\n data, errors = [], []\n for r in results:\n if isinstance(r, Exception):\n errors.append(str(r))\n else:\n data.append(r)\n\n if errors:\n print(f\"Errors: {errors}\")\n return data\n\n# Run from sync context\nresults = asyncio.run(fetch_all([\"https://api.example.com/a\", \"https://api.example.com/b\"]))\n```\n\n## Context Managers (Resource Safety)\n\n```python\nfrom contextlib import asynccontextmanager\n\n@asynccontextmanager\nasync def managed_session(api_key: str):\n client = httpx.AsyncClient(\n headers={\"Authorization\": f\"Bearer {api_key}\"},\n timeout=30.0,\n )\n try:\n yield client\n finally:\n await client.aclose()\n\n# Usage\nasync with managed_session(api_key) as client:\n data = await client.get(\"/endpoint\")\n```\n\n## Error Handling Pattern\n\n```python\nfrom enum import Enum\n\nclass ErrorCode(Enum):\n NOT_FOUND = \"not_found\"\n RATE_LIMITED = \"rate_limited\"\n SERVER_ERROR = \"server_error\"\n\nclass ApiError(Exception):\n def __init__(self, code: ErrorCode, message: str, retry_after: int | None = None):\n super().__init__(message)\n self.code = code\n self.retry_after = retry_after\n\nasync def call_api(endpoint: str) -> dict:\n async with httpx.AsyncClient() as client:\n try:\n resp = await client.get(endpoint)\n except httpx.TimeoutException:\n raise ApiError(ErrorCode.SERVER_ERROR, \"Request timed out\")\n\n if resp.status_code == 404:\n raise ApiError(ErrorCode.NOT_FOUND, f\"Not found: {endpoint}\")\n if resp.status_code == 429:\n retry_after = int(resp.headers.get(\"Retry-After\", \"60\"))\n raise ApiError(ErrorCode.RATE_LIMITED, \"Rate limited\", retry_after=retry_after)\n\n resp.raise_for_status()\n return resp.json()\n```\n\n## Retry with Exponential Backoff\n\n```python\nimport asyncio\nimport random\n\nasync def with_retry(\n coro_fn: Callable,\n max_attempts: int = 3,\n base_delay: float = 1.0,\n) -> T:\n for attempt in range(max_attempts):\n try:\n return await coro_fn()\n except ApiError as e:\n if e.code == ErrorCode.RATE_LIMITED:\n delay = e.retry_after or (base_delay * 2 ** attempt + random.random())\n await asyncio.sleep(delay)\n elif attempt == max_attempts - 1:\n raise\n raise RuntimeError(\"Unreachable\")\n```\n\n## Environment Variables (Safe Loading)\n\n```python\nimport os\nfrom dotenv import load_dotenv\n\nload_dotenv() # Loads .env file if present\n\ndef require_env(key: str) -> str:\n value = os.getenv(key)\n if not value:\n raise EnvironmentError(f\"Required env var not set: {key}\")\n return value\n\nOPENAI_API_KEY = require_env(\"OPENAI_API_KEY\")\nSUPABASE_URL = require_env(\"SUPABASE_URL\")\n```\n\n## Production Checklist\n\n- All public functions have type hints (mypy or pyright clean)\n- Async I/O for all network calls (no blocking `requests` in async code)\n- Context managers for all resource handles (files, HTTP clients, DB connections)\n- Retry logic on rate-limited endpoints\n- Secrets via `os.getenv()` — never hardcoded\n- Error types are specific, not bare `Exception`\n","html":"

Overview

\n

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

\n

Type Hints (Always Use)

\n
from typing import Optional, Union, TypeVar, Generic\nfrom collections.abc import Sequence, Callable, AsyncIterator\n\nT = TypeVar(\"T\")\n\ndef process_items(\n    items: Sequence[str],\n    transform: Callable[[str], str],\n    limit: Optional[int] = None,\n) -> list[str]:\n    results = [transform(item) for item in items]\n    return results[:limit] if limit else results\n\n# Python 3.10+ union syntax\ndef get_value(key: str) -> str | None:\n    return cache.get(key)\n
\n

Dataclasses for Structured Data

\n
from dataclasses import dataclass, field\nfrom datetime import datetime\n\n@dataclass\nclass ApiResponse:\n    status: int\n    data: dict\n    timestamp: datetime = field(default_factory=datetime.utcnow)\n    errors: list[str] = field(default_factory=list)\n\n    @property\n    def ok(self) -> bool:\n        return 200 <= self.status < 300\n
\n

Asyncio — Concurrent API Calls

\n
import asyncio\nimport httpx\n\nasync def fetch_one(client: httpx.AsyncClient, url: str) -> dict:\n    response = await client.get(url)\n    response.raise_for_status()\n    return response.json()\n\nasync def fetch_all(urls: list[str]) -> list[dict]:\n    async with httpx.AsyncClient(timeout=30.0) as client:\n        tasks = [fetch_one(client, url) for url in urls]\n        results = await asyncio.gather(*tasks, return_exceptions=True)\n\n    # Separate successes from failures\n    data, errors = [], []\n    for r in results:\n        if isinstance(r, Exception):\n            errors.append(str(r))\n        else:\n            data.append(r)\n\n    if errors:\n        print(f\"Errors: {errors}\")\n    return data\n\n# Run from sync context\nresults = asyncio.run(fetch_all([\"https://api.example.com/a\", \"https://api.example.com/b\"]))\n
\n

Context Managers (Resource Safety)

\n
from contextlib import asynccontextmanager\n\n@asynccontextmanager\nasync def managed_session(api_key: str):\n    client = httpx.AsyncClient(\n        headers={\"Authorization\": f\"Bearer {api_key}\"},\n        timeout=30.0,\n    )\n    try:\n        yield client\n    finally:\n        await client.aclose()\n\n# Usage\nasync with managed_session(api_key) as client:\n    data = await client.get(\"/endpoint\")\n
\n

Error Handling Pattern

\n
from enum import Enum\n\nclass ErrorCode(Enum):\n    NOT_FOUND = \"not_found\"\n    RATE_LIMITED = \"rate_limited\"\n    SERVER_ERROR = \"server_error\"\n\nclass ApiError(Exception):\n    def __init__(self, code: ErrorCode, message: str, retry_after: int | None = None):\n        super().__init__(message)\n        self.code = code\n        self.retry_after = retry_after\n\nasync def call_api(endpoint: str) -> dict:\n    async with httpx.AsyncClient() as client:\n        try:\n            resp = await client.get(endpoint)\n        except httpx.TimeoutException:\n            raise ApiError(ErrorCode.SERVER_ERROR, \"Request timed out\")\n\n        if resp.status_code == 404:\n            raise ApiError(ErrorCode.NOT_FOUND, f\"Not found: {endpoint}\")\n        if resp.status_code == 429:\n            retry_after = int(resp.headers.get(\"Retry-After\", \"60\"))\n            raise ApiError(ErrorCode.RATE_LIMITED, \"Rate limited\", retry_after=retry_after)\n\n        resp.raise_for_status()\n        return resp.json()\n
\n

Retry with Exponential Backoff

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

Environment Variables (Safe Loading)

\n
import os\nfrom dotenv import load_dotenv\n\nload_dotenv()  # Loads .env file if present\n\ndef require_env(key: str) -> str:\n    value = os.getenv(key)\n    if not value:\n        raise EnvironmentError(f\"Required env var not set: {key}\")\n    return value\n\nOPENAI_API_KEY = require_env(\"OPENAI_API_KEY\")\nSUPABASE_URL = require_env(\"SUPABASE_URL\")\n
\n

Production Checklist

\n\n"}