Chapter 25, Advanced
Integration with third-party APIs
Your API will call other APIs. Every network call can take time, fail, or return something unexpected, and what separates a fragile service from a robust one is how you handle each of these three cases.
The HTTP client
The original course uses the library requests, which is synchronous: within a route async def, it blocks the entire event loop (chapter 18). To call APIs from within a FastAPI application, the httpx has the same interface and also an asynchronous version. In this course, I use the httpx2, the same client as the TestClient uses:
uv add httpx2
The client should be created once, and not for each request: it keeps connections open and reuses them. The right place to create and close it is the lifespan, the code that runs when the application starts and stops:
from contextlib import asynccontextmanager
import httpx2
from fastapi import FastAPI, HTTPException, Request
from fastapi.testclient import TestClient
from pydantic import BaseModel, ValidationError
class Post(BaseModel):
id: int
title: str
def criar_app(transporte: httpx2.AsyncBaseTransport | None = None) -> FastAPI:
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.http = httpx2.AsyncClient(
base_url="https://jsonplaceholder.typicode.com",
timeout=httpx2.Timeout(5.0, connect=2.0),
transport=transporte,
)
yield
await app.state.http.aclose()
app = FastAPI(lifespan=lifespan)
@app.get("/posts/{post_id}", response_model=Post)
async def obter_post(post_id: int, request: Request):
try:
resposta = await request.app.state.http.get(f"/posts/{post_id}")
resposta.raise_for_status()
except httpx2.TimeoutException:
raise HTTPException(504, "A API externa demorou demais")
except httpx2.HTTPStatusError as erro:
if erro.response.status_code == 404:
raise HTTPException(404, "Post não encontrado")
raise HTTPException(502, "A API externa falhou")
except httpx2.TransportError:
raise HTTPException(502, "Não foi possível falar com a API externa")
try:
return Post.model_validate(resposta.json())
except ValidationError:
raise HTTPException(502, "A API externa devolveu um formato inesperado")
return app
Three decisions are in this code. The timeout is mandatory: without it, a stuck external API holds your request forever. Each failure becomes a status that describes the truth: 504 (the other end took too long), 502 (the other end failed or responded incorrectly) and 404 (does not exist), instead of a 500 generic one. And the external response is validated by Pydantic before being used: a third-party service can change the format without notice, and it's better to discover this in a 502 clear way than in a KeyError later.
Test without network
Tests should not depend on the internet (slow, unstable, with usage limits). The MockTransport client replaces the network with its own function, which decides what each address responds with:
def simulador(requisicao: httpx2.Request) -> httpx2.Response:
caminho = requisicao.url.path
if caminho == "/posts/1":
return httpx2.Response(200, json={"id": 1, "title": "primeiro", "extra": "ignorado"})
if caminho == "/posts/2":
return httpx2.Response(200, json={"id": 2})
if caminho == "/posts/3":
raise httpx2.ReadTimeout("lento demais", request=requisicao)
if caminho == "/posts/404":
return httpx2.Response(404)
return httpx2.Response(500)
app = criar_app(httpx2.MockTransport(simulador))
with TestClient(app) as cliente:
for numero in (1, 2, 3, 404, 9):
resposta = cliente.get(f"/posts/{numero}")
print(numero, resposta.status_code, resposta.json())
1 200 {'id': 1, 'title': 'primeiro'}
2 502 {'detail': 'A API externa devolveu um formato inesperado'}
3 504 {'detail': 'A API externa demorou demais'}
404 404 {'detail': 'Post não encontrado'}
9 502 {'detail': 'A API externa falhou'}
Each case was produced without network: success (with the extra field filtered), wrong format (502), timeout (504), the 404 and external server error. The with TestClient(app) is what triggers the lifespan: without it, the HTTP client does not exist.
Try again, only when it makes sense
Transient failures (a timeout, a 503) usually pass on the second attempt. Definitive failures (a 404, a 400) never pass, and repeating them is wasteful. The rule: repeat only the transient ones, with increasing wait between attempts and a limit:
import asyncio
async def get_com_tentativas(cliente, url: str, tentativas: int = 3, espera_base: float = 0.0):
for tentativa in range(1, tentativas + 1):
try:
resposta = await cliente.get(url)
if resposta.status_code < 500:
return resposta
except httpx2.TimeoutException:
if tentativa == tentativas:
raise
if tentativa < tentativas:
await asyncio.sleep(espera_base * 2 ** (tentativa - 1))
return resposta
chamadas = []
def instavel(requisicao: httpx2.Request) -> httpx2.Response:
chamadas.append(1)
if len(chamadas) < 3:
return httpx2.Response(503)
return httpx2.Response(200, json={"ok": True})
async def demonstrar():
async with httpx2.AsyncClient(transport=httpx2.MockTransport(instavel), base_url="http://x") as c:
resposta = await get_com_tentativas(c, "/dados")
return resposta.status_code, len(chamadas)
print(asyncio.run(demonstrar()))
(200, 3)
The simulated server failed twice and succeeded on the third. In production, espera_base it would be something like 0.5 seconds (the wait doubles with each attempt), and ideally with a bit of randomness to prevent many clients from trying again at the same moment.
External data is unreliable data
The response from an external API enters your application like any other input: validate the format, do not trust sizes or optional fields, and never pass the raw error from the other API to your client (it may contain internal addresses or keys). And keep API keys in environment variables (chapter 23), never in the code.
Exercise 1
Only the titles
Create GET /titulos, which calls GET /posts in the external API and returns only the list of titles. Use a MockTransport that returns three posts and checks the result.