Making HTTP Requests with httpx

httpx is Python’s modern HTTP client library — it supports both synchronous and asynchronous requests, has a clean API similar to the popular requests library, and is the standard choice for HTTP calls in FastAPI projects. When your FastAPI application needs to call other APIs — a payment gateway, an OAuth provider, an email service, or an external data source — httpx is the tool. The key distinction for FastAPI is async: FastAPI runs on an async event loop, and making synchronous HTTP calls inside async route handlers blocks the event loop. httpx.AsyncClient avoids this.

Synchronous Requests #

import httpx

# ── GET request ───────────────────────────────────────────────────────────────
response = httpx.get("https://api.github.com/users/tiangolo")
print(response.status_code)        # 200
print(response.headers["content-type"])  # application/json
data = response.json()             # parse JSON body
print(data["name"])                # Sebastián Ramírez

# ── With query parameters ─────────────────────────────────────────────────────
response = httpx.get(
    "https://api.github.com/search/repositories",
    params={"q": "fastapi", "sort": "stars", "per_page": 5}
)
# URL: .../search/repositories?q=fastapi&sort=stars&per_page=5

# ── POST with JSON body ────────────────────────────────────────────────────────
response = httpx.post(
    "https://api.example.com/auth/token",
    json={"username": "alice", "password": "secret"},  # auto-sets Content-Type: application/json
    headers={"Accept": "application/json"},
)

# ── PUT / PATCH / DELETE ──────────────────────────────────────────────────────
httpx.put("https://api.example.com/posts/1", json={"title": "Updated"})
httpx.patch("https://api.example.com/posts/1", json={"published": True})
httpx.delete("https://api.example.com/posts/1")

# ── With authentication ───────────────────────────────────────────────────────
token = "eyJhbGciOiJIUzI1NiJ9..."
response = httpx.get(
    "https://api.example.com/me",
    headers={"Authorization": f"Bearer {token}"}
)

Asynchronous Requests — For FastAPI #

import httpx

# ── Single async request ───────────────────────────────────────────────────────
async def fetch_user_from_github(username: str) -> dict:
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(f"https://api.github.com/users/{username}")
        response.raise_for_status()   # raises HTTPStatusError on 4xx/5xx
        return response.json()

# ── Reusable async client (application lifetime) ──────────────────────────────
# Better: create once, reuse across requests
class ExternalAPIClient:
    def __init__(self, base_url: str, api_key: str):
        self.client = httpx.AsyncClient(
            base_url = base_url,
            headers  = {"Authorization": f"Bearer {api_key}",
                        "Content-Type":  "application/json"},
            timeout  = httpx.Timeout(connect=5.0, read=30.0),
        )

    async def get(self, path: str, **kwargs) -> dict:
        response = await self.client.get(path, **kwargs)
        response.raise_for_status()
        return response.json()

    async def post(self, path: str, data: dict) -> dict:
        response = await self.client.post(path, json=data)
        response.raise_for_status()
        return response.json()

    async def close(self):
        await self.client.aclose()

# ── Parallel async requests ───────────────────────────────────────────────────
import asyncio

async def fetch_multiple(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks, return_exceptions=True)

        results = []
        for url, response in zip(urls, responses):
            if isinstance(response, Exception):
                results.append({"url": url, "error": str(response)})
            elif response.status_code == 200:
                results.append({"url": url, "data": response.json()})
            else:
                results.append({"url": url, "error": f"HTTP {response.status_code}"})
        return results

Error Handling with httpx #

import httpx

async def safe_get(url: str) -> dict | None:
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(url)

            # raise_for_status() raises HTTPStatusError for 4xx/5xx
            response.raise_for_status()
            return response.json()

    except httpx.TimeoutException:
        print(f"Request timed out: {url}")
        return None

    except httpx.ConnectError:
        print(f"Cannot connect to: {url}")
        return None

    except httpx.HTTPStatusError as e:
        print(f"HTTP error {e.response.status_code}: {url}")
        if e.response.status_code == 404:
            return None
        raise   # re-raise for 5xx errors

    except httpx.RequestError as e:
        # Base class for all connection-related errors
        print(f"Request failed: {e}")
        return None

# httpx exception hierarchy:
# httpx.RequestError          ← base for all request issues
#   httpx.ConnectError        ← cannot connect
#   httpx.TimeoutException    ← timeout
#     httpx.ConnectTimeout
#     httpx.ReadTimeout
#     httpx.WriteTimeout
# httpx.HTTPStatusError       ← 4xx or 5xx response (from raise_for_status())

httpx vs requests #

Common Mistakes #

Mistake 1 — Blocking async route with sync HTTP call #

❌ Wrong — blocks the event loop:

@app.get("/external")
async def get_external():
    data = httpx.get("https://api.example.com/data").json()   # BLOCKS event loop!
    return data

✅ Correct — use async client:

@app.get("/external")
async def get_external():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://api.example.com/data")
        return response.json()   # ✓ non-blocking

Mistake 2 — No timeout on external requests #

❌ Wrong — no timeout, blocks indefinitely:

async with httpx.AsyncClient() as client:   # default timeout: 5s, but can be None
    response = await client.get(url)   # could hang for minutes

✅ Correct — always set a timeout:

async with httpx.AsyncClient(timeout=10.0) as client:
    response = await client.get(url)   # raises TimeoutException after 10s ✓

Mistake 3 — Creating a new AsyncClient per request #

❌ Wrong — new client (new connection pool) created for every HTTP request:

@app.get("/posts/{id}")
async def get_post(id: int):
    async with httpx.AsyncClient() as client:   # new pool every request — inefficient
        ...

✅ Better — create one client at startup and share it via FastAPI state or dependency:

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http_client = httpx.AsyncClient(timeout=10.0)
    yield
    await app.state.http_client.aclose()   # ✓ shared, closed on shutdown

Quick Reference #