romankovsv/claude-code-python-devops-mlops/async-http-patterns

import httpx
# Sync
response = httpx.get("https://api.example.com/users")
response.raise_for_status()
data = response.json()
# Async
async with httpx.AsyncClient() as client:
    response = await client.get("https://api.example.com/users")
    response.raise_for_status()
    data = response.json()

import httpx
from contextlib import asynccontextmanager
class APIClient:
    def __init__(self, base_url: str, api_key: str):
        self.client = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=httpx.Timeout(30.0, connect=5.0),
            limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
        )
    async def get_user(self, user_id: int) -> dict:
        response = await self.client.get(f"/users/{user_id}")
        response.raise_for_status()
        return response.json()
    async def create_user(self, data: dict) -> dict:
        response = await self.client.post("/users", json=data)
        response.raise_for_status()
        return response.json()
    async def close(self):
        await self.client.aclose()

import asyncio
import httpx
from typing import TypeVar
T = TypeVar("T")
async def retry_request(
    client: httpx.AsyncClient,
    method: str,
    url: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
    **kwargs,
) -> httpx.Response:
    last_exc = None
    for attempt in range(max_retries + 1):
        try:
            response = await client.request(method, url, **kwargs)
            if response.status_code < 500:
                return response
            # Retry on 5xx
            last_exc = httpx.HTTPStatusError(
                f"Server error: {response.status_code}",
                request=response.request,
                response=response,
            )
        except (httpx.ConnectError, httpx.ReadTimeout) as exc:
            last_exc = exc
        if attempt < max_retries:
            delay = base_delay * (2 ** attempt)
            await asyncio.sleep(delay)
    raise last_exc

async def download_file(url: str, output_path: str):
    async with httpx.AsyncClient() as client:
        async with client.stream("GET", url) as response:
            response.raise_for_status()
            with open(output_path, "wb") as f:
                async for chunk in response.aiter_bytes(chunk_size=8192):
                    f.write(chunk)

import asyncio
import httpx
async def fetch_all(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        results = []
        for url, response in zip(urls, responses):
            if isinstance(response, Exception):
                results.append({"url": url, "error": str(response)})
            else:
                results.append({"url": url, "data": response.json()})
        return results

import asyncio
class RateLimitedClient:
    def __init__(self, client: httpx.AsyncClient, requests_per_second: float = 10):
        self.client = client
        self.semaphore = asyncio.Semaphore(int(requests_per_second))
        self.delay = 1.0 / requests_per_second
    async def get(self, url: str, **kwargs) -> httpx.Response:
        async with self.semaphore:
            response = await self.client.get(url, **kwargs)
            await asyncio.sleep(self.delay)
            return response

import aiohttp
async def fetch_with_aiohttp(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()
# With connection pooling
connector = aiohttp.TCPConnector(limit=100, limit_per_host=20)
session = aiohttp.ClientSession(connector=connector)

import respx
import httpx
import pytest
@respx.mock
@pytest.mark.asyncio
async def test_get_user():
    respx.get("https://api.example.com/users/1").mock(
        return_value=httpx.Response(200, json={"id": 1, "name": "Alice"})
    )
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/users/1")
        assert response.json()["name"] == "Alice"
@respx.mock
@pytest.mark.asyncio
async def test_api_error():
    respx.get("https://api.example.com/users/999").mock(
        return_value=httpx.Response(404, json={"error": "Not found"})
    )
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/users/999")
        assert response.status_code == 404