Calling External APIs from FastAPI

Most production FastAPI applications do not exist in isolation — they integrate with payment processors, OAuth providers, cloud storage APIs, notification services, and data providers. Calling external APIs from FastAPI route handlers requires careful design: async HTTP clients to avoid blocking, structured error handling for API failures, response validation with Pydantic models, and caching to avoid redundant calls that slow responses or exceed rate limits. In this lesson you will build a reusable external API integration pattern that is clean, testable, and resilient.

The External API Client Pattern #

# app/services/github_service.py
import httpx
from pydantic import BaseModel
from typing import Optional

# ── Pydantic model for the external API response ──────────────────────────────
class GitHubUser(BaseModel):
    login:      str
    name:       Optional[str] = None
    email:      Optional[str] = None
    public_repos: int
    followers:  int
    created_at: str

    model_config = {"extra": "ignore"}   # ignore unknown fields from the API

# ── Service class — wraps all interactions with GitHub API ────────────────────
class GitHubService:
    BASE_URL = "https://api.github.com"

    def __init__(self, client: httpx.AsyncClient):
        self.client = client   # injected — makes it testable

    async def get_user(self, username: str) -> GitHubUser:
        """Fetch a GitHub user's public profile."""
        try:
            response = await self.client.get(f"{self.BASE_URL}/users/{username}")
        except httpx.TimeoutException:
            raise ExternalAPIError("GitHub API timed out")
        except httpx.RequestError as e:
            raise ExternalAPIError(f"Cannot reach GitHub API: {e}")

        if response.status_code == 404:
            raise NotFoundError("GitHub user", username)
        if response.status_code == 403:
            raise ExternalAPIError("GitHub API rate limit exceeded")
        if response.status_code != 200:
            raise ExternalAPIError(f"GitHub API error: {response.status_code}")

        return GitHubUser.model_validate(response.json())

    async def get_repos(self, username: str, per_page: int = 10) -> list[dict]:
        response = await self.client.get(
            f"{self.BASE_URL}/users/{username}/repos",
            params={"per_page": per_page, "sort": "updated"},
        )
        response.raise_for_status()
        return [
            {"name": r["name"], "stars": r["stargazers_count"], "url": r["html_url"]}
            for r in response.json()
        ]

Integrating the Service into FastAPI #

# app/main.py — register shared HTTP client on lifespan
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
import httpx

from app.services.github_service import GitHubService
from app.exceptions import ExternalAPIError, NotFoundError

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Create shared client once — connection pooling
    app.state.http_client = httpx.AsyncClient(
        timeout = httpx.Timeout(connect=5.0, read=30.0),
        headers = {"Accept": "application/json",
                   "User-Agent": "MyFastAPIApp/1.0"},
    )
    yield
    await app.state.http_client.aclose()

app = FastAPI(lifespan=lifespan)

# Dependency — injects the service into route handlers
def get_github_service(request: Request) -> GitHubService:
    return GitHubService(client=request.app.state.http_client)

# Route handler — clean, no httpx code visible
@app.get("/github/{username}")
async def get_github_profile(
    username: str,
    github: GitHubService = Depends(get_github_service),
):
    user  = await github.get_user(username)
    repos = await github.get_repos(username, per_page=5)
    return {"profile": user.model_dump(), "top_repos": repos}

Caching External API Responses #

import time
from functools import lru_cache

# Simple in-memory cache with TTL
class TTLCache:
    def __init__(self, ttl_seconds: int = 300):
        self._cache: dict = {}
        self._ttl   = ttl_seconds

    def get(self, key: str):
        if key in self._cache:
            value, expires_at = self._cache[key]
            if time.time() < expires_at:
                return value
            del self._cache[key]   # expired
        return None

    def set(self, key: str, value):
        self._cache[key] = (value, time.time() + self._ttl)

    def invalidate(self, key: str):
        self._cache.pop(key, None)

# Cache GitHub user profiles for 5 minutes
_cache = TTLCache(ttl_seconds=300)

async def get_github_user_cached(username: str, service: GitHubService):
    cache_key = f"github:user:{username}"
    cached = _cache.get(cache_key)
    if cached:
        return cached

    user = await service.get_user(username)
    _cache.set(cache_key, user)
    return user

# For production: use Redis instead of in-memory
# pip install redis
# await redis_client.setex(f"github:user:{username}", 300, user.model_dump_json())

Handling Rate Limits #

import asyncio

async def get_with_retry(
    client: httpx.AsyncClient,
    url: str,
    max_retries: int = 3,
    backoff_seconds: float = 1.0,
) -> httpx.Response:
    """Retry on rate limit (429) or server errors (5xx) with exponential backoff."""
    for attempt in range(max_retries):
        try:
            response = await client.get(url)

            if response.status_code == 429:
                # Respect Retry-After header if present
                retry_after = int(response.headers.get("Retry-After", backoff_seconds))
                await asyncio.sleep(retry_after)
                continue

            if response.status_code >= 500 and attempt < max_retries - 1:
                await asyncio.sleep(backoff_seconds * (2 ** attempt))  # exponential backoff
                continue

            return response

        except httpx.TimeoutException:
            if attempt < max_retries - 1:
                await asyncio.sleep(backoff_seconds)
            else:
                raise

    raise ExternalAPIError(f"Max retries exceeded for {url}")

Common Mistakes #

Mistake 1 — Passing external API errors directly to clients #

❌ Wrong — exposing external API error details:

except httpx.HTTPStatusError as e:
    raise HTTPException(e.response.status_code, detail=e.response.text)
# Returns GitHub's raw error HTML/JSON to your API client

✅ Correct — map to your own error format:

except httpx.HTTPStatusError as e:
    logger.error(f"GitHub API error: {e.response.text}")
    raise ExternalAPIError("GitHub service unavailable")   # ✓ clean message

Mistake 2 — Using Pydantic without extra="ignore" for external APIs #

❌ Wrong — breaks when external API adds a new field:

class GitHubUser(BaseModel):
    login: str   # no extra="ignore" → ValidationError when new fields arrive

✅ Correct:

class GitHubUser(BaseModel):
    model_config = {"extra": "ignore"}   # ✓ future-proof
    login: str

Mistake 3 — No caching on frequently-called external APIs #

❌ Wrong — calling GitHub API on every request for the same user:

@app.get("/user/{username}")
async def show_user(username: str):
    return await github.get_user(username)   # GitHub called on every page load!

✅ Correct — cache responses with appropriate TTL:

return await get_github_user_cached(username, github)   # ✓ cached for 5 min

Quick Reference #