Python Cheatsheet 13 — httpx (Sync + Async HTTP Client)

uv add httpx

import httpx

r = httpx.get("https://api.example.com/users")
r.status_code             # 200
r.json()                  # dict
r.text                    # str
r.content                 # bytes
r.headers                 # dict-like
r.raise_for_status()      # raise on 4xx/5xx

import httpx

async with httpx.AsyncClient() as client:
    r = await client.get("https://api.example.com/users")
    print(r.json())

client = httpx.Client(
    base_url="https://api.example.com",
    timeout=10.0,
    headers={"Authorization": "Bearer ..."},
    follow_redirects=True,
)

r = client.get("/users")            # uses base_url
r = client.post("/users", json={"name": "x"})
client.close()

# Or:
with httpx.Client(...) as client:
    r = client.get(...)

@asynccontextmanager
async def lifespan(app):
    app.state.http = httpx.AsyncClient(timeout=10.0)
    yield
    await app.state.http.aclose()

# In handler
@app.get("/proxy")
async def proxy(request: Request):
    r = await request.app.state.http.get("https://api.example.com/data")
    return r.json()

r = client.get(url, params={"q": "foo"})
r = client.post(url, json={"x": 1})
r = client.post(url, data={"x": 1})           # form
r = client.post(url, content=b"raw bytes")
r = client.put(url, json={...})
r = client.patch(url, json={...})
r = client.delete(url)
r = client.head(url)
r = client.options(url)

client.get(url, headers={"X-Custom": "v"})
client.get(url, cookies={"session": "..."})

# Persistent cookies
client = httpx.Client()
client.get("https://example.com/login", params={"u": "x"})
# Cookies stored; sent on next request automatically

# Basic
client.get(url, auth=("user", "pass"))

# Bearer
client.get(url, headers={"Authorization": "Bearer abc"})

# Custom auth class
class MyAuth(httpx.Auth):
    def auth_flow(self, request):
        request.headers["X-API-Key"] = self.key
        yield request

client.get(url, auth=MyAuth())

# Total
httpx.get(url, timeout=10.0)

# Granular
httpx.get(url, timeout=httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0))

# Disable
httpx.get(url, timeout=None)

transport = httpx.HTTPTransport(retries=3)
client = httpx.Client(transport=transport)

transport = httpx.AsyncHTTPTransport(retries=3)
client = httpx.AsyncClient(transport=transport)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    retry=retry_if_exception_type((httpx.HTTPError, httpx.RequestError)),
    stop=stop_after_attempt(5),
    wait=wait_exponential(min=1, max=10),
)
async def fetch_with_retry(url: str) -> dict:
    async with httpx.AsyncClient() as c:
        r = await c.get(url, timeout=10)
        r.raise_for_status()
        return r.json()

with httpx.stream("GET", url) as r:
    for chunk in r.iter_bytes(chunk_size=8192):
        process(chunk)

# Lines
with httpx.stream("GET", url) as r:
    for line in r.iter_lines():
        ...

# JSON SSE
with httpx.stream("GET", url) as r:
    for line in r.iter_lines():
        if line.startswith("data:"):
            data = json.loads(line[5:].strip())
            yield data

async with httpx.AsyncClient() as c:
    async with c.stream("GET", url) as r:
        async for chunk in r.aiter_bytes():
            ...

files = {"file": open("data.csv", "rb")}
r = client.post(url, files=files)

# Multipart with additional fields
data = {"name": "alice"}
r = client.post(url, files=files, data=data)

client = httpx.Client(proxies="http://proxy:8080")
# Or per-scheme
client = httpx.Client(proxies={
    "http://": "http://proxy:8080",
    "https://": "http://proxy:8080",
})

httpx.get(url, verify="/path/to/ca.pem")
httpx.get(url, verify=False)      # disable; testing only!

client = httpx.Client(http2=True)

async with httpx.AsyncClient() as c:
    tasks = [c.get(url) for url in urls]
    responses = await asyncio.gather(*tasks)

sem = asyncio.Semaphore(10)
async def fetch(url):
    async with sem:
        return await c.get(url)

try:
    r = client.get(url, timeout=5)
    r.raise_for_status()
except httpx.TimeoutException: ...
except httpx.HTTPStatusError as e: ...      # 4xx/5xx
except httpx.RequestError as e: ...         # network errors

def log_request(request):
    print(f"-> {request.method} {request.url}")

def log_response(response):
    response.read()
    print(f"<- {response.status_code}")

client = httpx.Client(event_hooks={
    "request": [log_request],
    "response": [log_response],
})