Python httpx

Modern HTTP client with async support, connection pooling, and HTTP/2.

Basic Requests

GET, POST, PUT, DELETE — httpx mirrors requests API but adds async support

import httpx

# GET with query parameters
response = httpx.get(
    "https://api.example.com/devices",
    params={"vlan": 10, "limit": 50},
)

# POST with JSON body
response = httpx.post(
    "https://api.example.com/devices",
    json={"hostname": "switch-01", "ip": "10.50.1.100", "vlan": 10},
)

# PUT and DELETE
response = httpx.put("https://api.example.com/devices/1", json={"vlan": 20})
response = httpx.delete("https://api.example.com/devices/1")

Response Handling

Status code, JSON parsing, and raise_for_status() for error propagation

import httpx

response = httpx.get("https://api.example.com/devices/1")

response.status_code        # 200
response.json()             # parsed dict
response.text               # raw string body
response.headers            # case-insensitive dict
response.headers["content-type"]  # "application/json"

# Raise httpx.HTTPStatusError for 4xx/5xx -- do this instead of manual status checks
response.raise_for_status()

# Check without raising
response.is_success         # True for 2xx
response.is_redirect        # True for 3xx
response.is_client_error    # True for 4xx
response.is_server_error    # True for 5xx

Headers and Authentication

Custom headers, bearer tokens, and basic auth

import httpx

# Custom headers
response = httpx.get(
    "https://api.example.com/secure",
    headers={"X-API-Key": api_key, "Accept": "application/json"},
)

# Bearer token auth
response = httpx.get(
    "https://api.example.com/secure",
    headers={"Authorization": f"Bearer {token}"},
)

# Basic auth -- httpx handles encoding
response = httpx.get(
    "https://api.example.com/secure",
    auth=("username", "password"),
)

Timeouts

Always set timeouts — httpx defaults to 5s, but be explicit in production

import httpx

# Single timeout for all phases
response = httpx.get("https://api.example.com/slow", timeout=30.0)

# Granular timeouts -- connect fast, read slow (large responses)
timeout = httpx.Timeout(
    connect=5.0,
    read=30.0,
    write=10.0,
    pool=5.0,
)
response = httpx.get("https://api.example.com/report", timeout=timeout)

Synchronous Client (Session)

Client reuses TCP connections — faster for multiple requests to same host

import httpx

with httpx.Client(
    base_url="https://api.example.com",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
) as client:
    devices = client.get("/devices").json()
    for device in devices:
        detail = client.get(f"/devices/{device['id']}").json()
        print(detail["hostname"])

Async Client

AsyncClient for FastAPI, background tasks, and concurrent requests

import asyncio
import httpx

async def fetch_all_devices():
    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        headers={"Authorization": f"Bearer {token}"},
    ) as client:
        # Sequential
        response = await client.get("/devices")
        devices = response.json()

        # Concurrent -- fan out requests
        tasks = [client.get(f"/devices/{d['id']}") for d in devices]
        responses = await asyncio.gather(*tasks)
        return [r.json() for r in responses]

Streaming Responses

Stream large responses without loading entire body into memory

import httpx

with httpx.stream("GET", "https://example.com/large-export.csv") as response:
    response.raise_for_status()
    with open("export.csv", "wb") as f:
        for chunk in response.iter_bytes(chunk_size=8192):
            f.write(chunk)

# Async streaming
async with httpx.AsyncClient() as client:
    async with client.stream("GET", url) as response:
        async for line in response.aiter_lines():
            process(line)

File Upload

Multipart file upload — single and multiple files

import httpx
from pathlib import Path

# Single file
files = {"file": ("config.txt", Path("config.txt").read_bytes(), "text/plain")}
response = httpx.post("https://api.example.com/upload", files=files)

# File with additional form fields
response = httpx.post(
    "https://api.example.com/upload",
    files={"file": ("backup.tar.gz", open("backup.tar.gz", "rb"))},
    data={"description": "Daily ISE backup", "category": "backup"},
)

Error Handling Pattern

Structured error handling — distinguish network errors from HTTP errors

import httpx

try:
    response = httpx.get("https://api.example.com/devices", timeout=10.0)
    response.raise_for_status()
    data = response.json()
except httpx.ConnectError:
    print("Cannot reach API -- check network/DNS")
except httpx.TimeoutException:
    print("Request timed out -- API may be overloaded")
except httpx.HTTPStatusError as e:
    print(f"HTTP {e.response.status_code}: {e.response.text}")
except httpx.RequestError as e:
    print(f"Request failed: {e}")