HTTPX Snippets: ein praktischer Leitfaden

1 Einführung in HTTPX

HTTPX ist der moderne HTTP-Client für Python. Er bewahrt die vertraute requests API und bietet gleichzeitig HTTP/2, asynchrone Unterstützung und Connection Pooling out of the box. Wenn Sie requests kennen, kennen Sie bereits den Großteil von HTTPX – fügen Sie Ihrer mentalen Checkliste nur noch „sync or async?“ hinzu.

Synchron und asynchron mit dem gleichen mentalen Modell
Moderne Protokolle: HTTP/2, wenn verfügbar
Gute Standardeinstellungen: Connection Pooling, Timeouts, Retries

Klicken Sie auf „Weiter“, um HTTPX zu installieren und Ihre erste Anfrage zu senden.

2 Installation

Verwenden Sie eine virtuelle Umgebung – und installieren Sie dann HTTPX. Der [http]-Extra holt den Standardtransport (keine zusätzlichen Abhängigkeiten erforderlich).

pip install httpx[http]

Wenn Sie HTTP/2 benötigen, installieren Sie auch h2: pip install h2. Für HTTPS unter macOS benötigen Sie möglicherweise Zertifikate: pip install certifi und setzen Sie SSL_CERT_FILE oder REQUESTS_CA_BUNDLE.

3 Ein einfaches GET

Hier ist ein minimales synchrone GET. Es gibt den Status und einen kurzen Auszug aus.

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])

Der Client wird als Context Manager verwendet, sodass Verbindungen sauber freigegeben werden. Allein das verhindert eine überraschende Anzahl von „zu viele offene Dateien“-Fehlern.

4 Header, Parameter und Timeouts

Übergeben Sie Header und Query-Parameter als normale Dictionaries. Setzen Sie immer explizite 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"))

Verwenden Sie eine base_url, um Duplizierung zu reduzieren, und r.raise_for_status(), um HTTP-Fehler in Ausnahmen zu verwandeln, anstatt sie stillschweigend zu ignorieren.

5 POST mit JSON

JSON POSTs sind eine Einzeiler. Der Client serialisiert und setzt den korrekten 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 Fehlerbehandlung

Verwenden Sie spezifische Ausnahmen, wo immer möglich; greifen Sie andernfalls auf die Basisklasse zurück.

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("Konnte keine Verbindung herstellen (Netzwerk-Ausfall, DNS, abgelehnt).")
except ReadTimeout:
    print("Server hat zu lange zur Antwort gebraucht.")
except RemoteProtocolError:
    print("Verbindung mitten im Flug abgebrochen.")
except httpx.HTTPStatusError as e:
    print(f"HTTP-Fehler: {e.response.status_code} {e.response.reason_phrase}")
except httpx.RequestError as e:
    print(f"Anfrage fehlgeschlagen: {e}")

7 Asynchrone Anfragen (asyncio)

Wechseln Sie zu httpx.AsyncClient, wenn Sie Parallelität benötigen. Gruppieren Sie I/O-Aufrufe mit 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 Wiederholungsversuche und Backoff

Bei transienten Fehlern umschließen Sie den Transport mit einer Wiederholungsrichtlinie unter Verwendung von httpx.Limits und 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)

Hinweis: Die Wiederholungslogik liegt in der Transportschicht. Seien Sie vorsichtig – Wiederholungsversuche können Ausfälle verstärken.

9 Erweiterte Timeouts

Teilen Sie Verbindungs- und Lese-Timeouts für eine präzisere Kontrolle auf. Sie können auch den Lese-Timeout (None) für Streaming-Antworten deaktivieren.

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 Dateien hochladen (Multipart)

Senden Sie Dateien mit files= und reguläre Formularfelder mit 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 HTTPX in FastAPI verwenden

Erstellen Sie einen einzigen, wiederverwendbaren Client beim Start der Anwendung; schließen Sie ihn beim Herunterfahren.

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()

Produktion: Montieren Sie Limits und eine Wiederholungsrichtlinie auf dem Transport. Cache für gängige Antworten, wenn der Upstream gesprächig ist.