Capítulo 25, Avançado
Integração com APIs de terceiros
Sua API vai chamar outras APIs. Toda chamada de rede pode demorar, falhar ou devolver algo inesperado, e o que separa um serviço frágil de um robusto é como você trata cada um desses três casos.
O cliente HTTP
O curso original usa a biblioteca requests, que é síncrona: dentro de uma rota async def, ela bloqueia o laço de eventos inteiro (capítulo 18). Para chamar APIs de dentro de uma aplicação FastAPI, o httpx tem a mesma interface e também uma versão assíncrona. Neste curso uso o httpx2, o mesmo cliente que o TestClient usa:
uv add httpx2
O cliente deve ser criado uma vez, e não a cada requisição: ele mantém conexões abertas e as reaproveita. O lugar certo para criá-lo e fechá-lo é o lifespan, o código que roda quando a aplicação liga e desliga:
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
Três decisões estão nesse código. O timeout é obrigatório: sem ele, uma API externa travada prende a sua requisição para sempre. Cada falha vira um status que descreve a verdade: 504 (a outra ponta demorou), 502 (a outra ponta falhou ou respondeu errado) e 404 (não existe), em vez de um 500 genérico. E a resposta externa é validada pelo Pydantic antes de ser usada: um serviço de terceiros pode mudar o formato sem avisar, e é melhor descobrir isso em um 502 claro do que em um KeyError adiante.
Testar sem rede
Testes não devem depender da internet (lenta, instável, com limites de uso). O MockTransport do cliente substitui a rede por uma função sua, que decide o que cada endereço responde:
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'}
Cada caso foi produzido sem rede: o sucesso (com o campo extra filtrado), o formato errado (502), o timeout (504), o 404 e o erro do servidor externo. O with TestClient(app) é o que dispara o lifespan: sem ele, o cliente HTTP não existe.
Tentar de novo, só quando faz sentido
Falhas transitórias (um timeout, um 503) costumam passar na segunda tentativa. Falhas definitivas (um 404, um 400) nunca passam, e repeti-las é desperdício. A regra: repetir só o transitório, com espera crescente entre as tentativas e um limite:
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)
O servidor simulado falhou duas vezes e acertou na terceira. Em produção, espera_base seria algo como 0,5 segundo (a espera dobra a cada tentativa), e idealmente com um pouco de aleatoriedade para evitar que muitos clientes tentem de novo no mesmo instante.
Dados de fora são dados não confiáveis
A resposta de uma API externa entra na sua aplicação como qualquer outra entrada: valide o formato, não confie em tamanhos nem em campos opcionais, e nunca repasse ao seu cliente o erro cru da outra API (pode conter endereços internos ou chaves). E guarde as chaves de API em variáveis de ambiente (capítulo 23), nunca no código.
Exercício 1
Só os títulos
Crie GET /titulos, que chame GET /posts na API externa e devolva só a lista de títulos. Use um MockTransport que devolva três posts e confira o resultado.