httpx API Patterns

httpx patterns for Python API clients — sync and async.

Basic Requests

GET with headers — httpx replaces requests for async support
import httpx

response = httpx.get(
    "https://api.example.com/v1/devices",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
)
response.raise_for_status()
devices = response.json()
POST JSON payload
response = httpx.post(
    "https://api.example.com/v1/devices",
    headers={"Authorization": f"Bearer {token}"},
    json={"name": "sw-core-01", "vlan": 10},
    timeout=10.0,
)
print(response.status_code, response.json())
PUT, PATCH, DELETE — same pattern
# Full replacement
httpx.put(url, json=full_payload)

# Partial update
httpx.patch(url, json={"status": "active"})

# Delete
httpx.delete(url)

Client Sessions

Reuse connections — httpx.Client for multiple requests
with httpx.Client(
    base_url="https://api.example.com",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
) as client:
    devices = client.get("/v1/devices").json()
    for device in devices:
        detail = client.get(f"/v1/devices/{device['id']}").json()
        print(detail["name"], detail["status"])
httpx.Client reuses TCP connections (HTTP keep-alive), reducing latency for sequential calls to the same host.

Async Requests

AsyncClient — for FastAPI routes calling other APIs
import httpx

async def fetch_devices() -> list[dict]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": f"Bearer {token}"},
        timeout=10.0,
    ) as client:
        response = await client.get("/v1/devices")
        response.raise_for_status()
        return response.json()
Parallel async requests — asyncio.gather for concurrent calls
import asyncio
import httpx

async def fetch_all_details(device_ids: list[str]) -> list[dict]:
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": f"Bearer {token}"},
    ) as client:
        tasks = [client.get(f"/v1/devices/{did}") for did in device_ids]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses if r.status_code == 200]

Authentication Patterns

Basic auth — ISE ERS via httpx
response = httpx.get(
    "https://ise-01.inside.domusdigitalis.dev:9060/ers/config/endpoint",
    auth=("admin", ise_password),
    headers={"Accept": "application/json"},
    verify=False,  # self-signed cert in lab
    timeout=15.0,
)
Bearer token with automatic refresh
import httpx

class TokenAuth(httpx.Auth):
    def __init__(self, token: str):
        self.token = token

    def auth_flow(self, request):
        request.headers["Authorization"] = f"Bearer {self.token}"
        yield request

client = httpx.Client(auth=TokenAuth(token), base_url="https://api.example.com")
Client certificate (mTLS)
client = httpx.Client(
    cert=("/path/to/client.pem", "/path/to/client.key"),
    verify="/path/to/ca-chain.pem",
)

Error Handling

Structured error handling
try:
    response = client.get("/v1/devices/42")
    response.raise_for_status()
except httpx.HTTPStatusError as e:
    match e.response.status_code:
        case 401:
            print("Token expired or invalid")
        case 404:
            print("Device not found")
        case 429:
            retry = int(e.response.headers.get("Retry-After", 60))
            print(f"Rate limited -- wait {retry}s")
        case _:
            print(f"HTTP {e.response.status_code}: {e.response.text}")
except httpx.ConnectError:
    print("Cannot connect to API")
except httpx.TimeoutException:
    print("Request timed out")

Response Inspection

Debug response metadata
response = client.get("/v1/devices")
print(f"Status: {response.status_code}")
print(f"Content-Type: {response.headers['content-type']}")
print(f"Elapsed: {response.elapsed.total_seconds():.3f}s")
print(f"URL: {response.url}")
print(f"Redirected: {response.is_redirect}")

Timeout Configuration

Granular timeouts — connect vs read vs write
timeout = httpx.Timeout(
    connect=5.0,    # TCP handshake
    read=30.0,      # waiting for response body
    write=10.0,     # sending request body
    pool=5.0,       # waiting for connection from pool
)

client = httpx.Client(timeout=timeout)

File Upload

Multipart file upload
with open("backup.tar.gz", "rb") as f:
    response = client.post(
        "/v1/upload",
        files={"file": ("backup.tar.gz", f, "application/gzip")},
        data={"description": "Config backup"},
    )

httpx vs requests

Feature              httpx                    requests
Async support        Yes (AsyncClient)        No (needs aiohttp)
HTTP/2               Yes (h2 extra)           No
Connection pool      Built-in                 Session-based
Timeout              Granular (4 types)       Single value
Type hints           Full                     Partial
API compatibility    Nearly identical          --
httpx is the successor to requests for new projects. Use it in domus-api and any new Python tooling.