FastAPI Working with External APIs

Most real applications call other services — a payment gateway, a weather API, a mapping service, or a social login provider. HTTPx is the recommended HTTP client for FastAPI because it supports both synchronous and asynchronous requests, making it a perfect match for async FastAPI routes.

Install HTTPx

pip install httpx

Synchronous HTTPx (Quick and Simple)

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/weather/{city}")
def get_weather(city: str):
    response = httpx.get(f"https://wttr.in/{city}?format=j1")
    response.raise_for_status()   ← raises exception for 4xx/5xx
    return response.json()

Async HTTPx (Recommended for FastAPI)

@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://wttr.in/{city}?format=j1")
        response.raise_for_status()
        return response.json()

async with httpx.AsyncClient() as client:
  │
  └── opens a client → makes the request → closes the client
      automatically (even if an error occurs)

Reusing the Client Across Requests

Creating and destroying an HTTP client on every request is wasteful. Share one client for the whole application lifecycle using FastAPI's startup and shutdown events:

from contextlib import asynccontextmanager

http_client: httpx.AsyncClient = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global http_client
    http_client = httpx.AsyncClient()   ← created once at startup
    yield
    await http_client.aclose()          ← closed once at shutdown

app = FastAPI(lifespan=lifespan)

@app.get("/github/{username}")
async def github_profile(username: str):
    response = await http_client.get(
        f"https://api.github.com/users/{username}"
    )
    return response.json()

Connection pool lifecycle:
  App starts → AsyncClient created (connection pool ready)
  Request 1  → reuses existing connection → fast
  Request 2  → reuses existing connection → fast
  App stops  → AsyncClient closed cleanly

Sending Headers and Auth Tokens

@app.get("/protected-resource")
async def fetch_protected():
    headers = {
        "Authorization": "Bearer my-api-token",
        "Accept": "application/json"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(
            "https://api.example.com/data",
            headers=headers
        )
        return response.json()

Sending POST with JSON Body

@app.post("/notify")
async def send_notification(message: str):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://hooks.slack.com/services/YOUR/HOOK/URL",
            json={"text": message}
        )
        return {"slack_status": response.status_code}

Handling External API Errors Gracefully

from fastapi import HTTPException

@app.get("/exchange-rate/{currency}")
async def get_rate(currency: str):
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                f"https://api.exchangerate.host/latest?base={currency}",
                timeout=5.0   ← fail fast if slow
            )
            response.raise_for_status()
            return response.json()
        except httpx.TimeoutException:
            raise HTTPException(503, detail="Exchange rate service timed out")
        except httpx.HTTPStatusError as e:
            raise HTTPException(502, detail=f"Upstream error: {e.response.status_code}")

Error types to catch:
  httpx.TimeoutException     → service took too long
  httpx.ConnectError         → could not reach the server
  httpx.HTTPStatusError      → got a 4xx or 5xx response
  httpx.RequestError         → general request failure

Setting Default Timeouts

# Timeout breakdown:
timeout = httpx.Timeout(
    connect=3.0,   ← max seconds to establish connection
    read=10.0,     ← max seconds to receive response body
    write=5.0,     ← max seconds to send request body
    pool=1.0       ← max seconds waiting for a connection from pool
)

client = httpx.AsyncClient(timeout=timeout)

Key Points

• HTTPx supports both sync and async requests — use async inside async def routes.
• Use async with httpx.AsyncClient() to open and close the client safely.
• Share one AsyncClient across all requests using the app lifespan for better performance.
• Call response.raise_for_status() to automatically raise an error on 4xx/5xx responses.
• Always set a timeout so your route does not hang forever waiting for a slow external service.