HTTPX Snippets 实用指南

1 Introduction to HTTPX

HTTPX is the modern HTTP client for Python. It preserves the familiar requests API while giving you HTTP/2, async support, and connection pooling out of the box. If you know requests, you already know most of HTTPX—just add “sync or async?” to your mental checklist.

Sync and async with the same mental model
Modern protocols: HTTP/2 when available
Great defaults: connection pooling, timeouts, retries

Click “Next” to install HTTPX and send your first request.

2 Installation

Use a virtual environment—then install HTTPX. The [http] extra pulls in the standard transport (no extra dependencies required).

pip install httpx[http]

If you need HTTP/2, also install h2: pip install h2. For HTTPS under macOS, you may need certificates: pip install certifi and set SSL_CERT_FILE or REQUESTS_CA_BUNDLE.

3 A Simple GET

Here’s a minimal sync GET. It prints status and a short excerpt.

import httpx

with httpx.Client() as client:
    r = client.get("https://httpbin.org/get")
    print("status:", r.status_code)
    print("content[:120]:", r.text[:120])

The client is used as a context manager so connections are cleanly released. That alone prevents a surprising number of “too many open files” errors.

4 Headers, Params, and Timeouts

Pass headers and query params as plain dicts. Always set explicit timeouts.

import httpx

with httpx.Client(
    base_url="https://httpbin.org",
    headers={"User-Agent": "httpx-snippets/1.0"},
    params={"page": "1", "per_page": "5"},
    timeout=5.0,
) as client:
    r = client.get("/get")
    r.raise_for_status()
    data = r.json()
    print(data.get("args"))

Use a base_url to reduce duplication and r.raise_for_status() to turn HTTP errors into exceptions instead of silent fails.

5 POST with JSON

JSON POSTs are a one-liner. The client serializes and sets the correct Content-Type.

import httpx

payload = {"name": "Ada", "skills": ["math", "coding"]}

with httpx.Client(base_url="https://httpbin.org") as client:
    r = client.post("/post", json=payload)
    r.raise_for_status()
    print(r.json()["json"])

6 Error Handling

Use specific exceptions where you can; fall back to the base class otherwise.

import httpx
from httpx import ConnectTimeout, ReadTimeout, RemoteProtocolError

try:
    with httpx.Client(timeout=1.0) as client:
        r = client.get("https://httpbin.org/delay/2")
        r.raise_for_status()
except ConnectTimeout:
    print("Could not connect (network down, DNS, refused).")
except ReadTimeout:
    print("Server took too long to respond.")
except RemoteProtocolError:
    print("Connection dropped mid-flight.")
except httpx.HTTPStatusError as e:
    print(f"HTTP error: {e.response.status_code} {e.response.reason_phrase}")
except httpx.RequestError as e:
    print(f"Request failed: {e}")

7 Async Requests (asyncio)

Switch to httpx.AsyncClient when you need concurrency. Group I/O calls with asyncio.gather.

import asyncio
import httpx

async def fetch(url: str):
    async with httpx.AsyncClient(timeout=5.0) as client:
        r = await client.get(url)
        r.raise_for_status()
        return r.json()

async def main():
    urls = ["https://httpbin.org/get"] * 4
    results = await asyncio.gather(*(fetch(u) for u in urls))
    for i, data in enumerate(results, 1):
        print(i, data.get("origin"))

asyncio.run(main())

8 Retries and Backoff

For transient failures, wrap the transport with a retry policy using httpx.Limits and Timeout.

import httpx
from urllib3.util.retry import Retry
from httpx import LegacyHTTPTransport

retry = Retry(
    total=3,
    backoff_factor=0.5,
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS"],
)

limits = httpx.Limits(max_keepalive_connections=5, max_connections=10)
timeout = httpx.Timeout(5.0, connect=2.0)

transport = LegacyHTTPTransport(retries=retry)
client = httpx.Client(transport=transport, limits=limits, timeout=timeout)

r = client.get("https://httpbin.org/status/503")
print(r.status_code)

Note: retry logic lives in the transport layer. Keep it conservative— retries can amplify outages.

9 Advanced Timeouts

Split connect and read timeouts for tighter control. You can also disable read timeout (None) for streaming responses.

import httpx

timeout = httpx.Timeout(connect=2.0, read=5.0, write=2.0, pool=1.0)
with httpx.Client(timeout=timeout) as client:
    r = client.get("https://httpbin.org/stream/3")  # stream lines
    for line in r.iter_lines():
        print(line)

10 Uploading Files (Multipart)

Send files with files= and regular form fields with data=.

import httpx

with httpx.Client(base_url="https://httpbin.org") as client:
    with open("README.md", "rb") as f:
        files = {"file": ("README.md", f, "text/markdown")}
        data = {"uploader": "me"}
        r = client.post("/post", files=files, data=data)
        r.raise_for_status()
        print(r.json()["headers"].get("Content-Type"))

11 Using HTTPX in FastAPI

Create a single, reusable client on app startup; close it on shutdown.

from fastapi import FastAPI
import httpx

app = FastAPI()
client = httpx.Client(base_url="https://httpbin.org", timeout=5.0)

@app.on_event("startup")
async def on_startup():
    await client.__aenter__()

@app.on_event("shutdown")
async def on_shutdown():
    await client.__aexit__(None, None, None)

@app.get("/proxy/get")
async def proxy_get():
    r = await client.get("/get")
    r.raise_for_status()
    return r.json()

In production, mount Limits and a retry policy on the transport. Cache common responses if the upstream is chatty.