HTTPX em Python: timeouts, retries e clientes resilientes

Consumir APIs externas parece simples no primeiro script: fazer um GET, ler o JSON e seguir em frente. O problema aparece quando esse script vira uma rotina diária, um endpoint de produção, uma integração com CRM, um robô de cobrança ou um pipeline de dados. A API demora, cai, devolve 429, muda o formato de erro, mantém conexão aberta por tempo demais ou falha exatamente quando você mais precisa da resposta.

Em Python moderno, HTTPX é uma das melhores opções para escrever clientes HTTP porque oferece API síncrona e assíncrona, suporte a HTTP/2, configuração explícita de timeout, pool de conexões e integração boa com testes. Mas a biblioteca não elimina decisões de arquitetura: você ainda precisa definir quanto tempo esperar, quando tentar de novo, como registrar falhas e como impedir que uma API instável derrube seu sistema.

Este guia complementa os conteúdos de Python e APIs: consumindo dados, APIs REST com FastAPI, logging em Python e testes com pytest. O foco aqui é produção: clientes pequenos, previsíveis e fáceis de testar.

Por que timeout explícito é obrigatório

O erro mais comum em integrações Python é chamar uma API sem timeout. Em desenvolvimento, tudo funciona. Em produção, uma conexão lenta pode prender workers, atrasar filas, deixar usuários esperando e criar efeito dominó. Timeout não é detalhe; é limite de dano.

Com HTTPX, você pode definir timeout simples:

import httpx

response = httpx.get("https://api.example.com/status", timeout=5.0)
response.raise_for_status()
print(response.json())

Esse formato já é melhor do que não ter limite, mas projetos profissionais costumam separar fases da requisição:

import httpx

timeout = httpx.Timeout(
    connect=2.0,  # tempo para abrir conexão
    read=5.0,     # tempo entre bytes da resposta
    write=5.0,    # tempo para enviar corpo
    pool=2.0,     # espera por conexão livre no pool
)

with httpx.Client(timeout=timeout) as client:
    response = client.get("https://api.example.com/clientes")
    response.raise_for_status()

Essa diferença importa quando o serviço recebe muitas chamadas. Um connect curto evita ficar preso tentando abrir conexão com host indisponível. Um pool curto mostra que sua própria aplicação está sem conexões disponíveis. Um read maior permite respostas um pouco pesadas sem abrir mão de limite.

Use Client em vez de chamadas soltas

Para scripts rápidos, httpx.get() resolve. Para integração real, prefira httpx.Client. Ele reaproveita conexões, centraliza headers, timeout, URL base e hooks. Também deixa o código mais fácil de testar porque você injeta o cliente na função.

import httpx

class ClienteCRM:
    def __init__(self, base_url: str, token: str) -> None:
        self._client = httpx.Client(
            base_url=base_url,
            timeout=httpx.Timeout(connect=2.0, read=5.0, write=5.0, pool=2.0),
            headers={"Authorization": f"Bearer {token}"},
        )

    def buscar_cliente(self, cliente_id: str) -> dict:
        response = self._client.get(f"/clientes/{cliente_id}")
        response.raise_for_status()
        return response.json()

    def close(self) -> None:
        self._client.close()

Em aplicações FastAPI, você pode criar o cliente no ciclo de vida da aplicação e fechá-lo no shutdown. Em scripts, use context manager. Em jobs de fila, crie o cliente por processo ou por lote, não a cada item individual, para evitar overhead desnecessário.

Tratamento de erro: separe falhas esperadas

Nem todo erro HTTP significa a mesma coisa. Um 404 pode indicar cliente inexistente. Um 401 indica credencial inválida. Um 429 pede redução de ritmo. Um 500 pode ser instabilidade temporária. Já ConnectTimeout e ReadTimeout são falhas de rede/latência, não respostas da API.

Um cliente legível traduz esses casos para exceções do seu domínio:

import httpx

class ClienteNaoEncontrado(Exception):
    pass

class CRMIndisponivel(Exception):
    pass


def buscar_cliente(client: httpx.Client, cliente_id: str) -> dict:
    try:
        response = client.get(f"/clientes/{cliente_id}")
        if response.status_code == 404:
            raise ClienteNaoEncontrado(cliente_id)
        response.raise_for_status()
        return response.json()
    except (httpx.TimeoutException, httpx.NetworkError) as exc:
        raise CRMIndisponivel("Falha de rede ao consultar CRM") from exc
    except httpx.HTTPStatusError as exc:
        if 500 <= exc.response.status_code < 600:
            raise CRMIndisponivel("CRM retornou erro temporário") from exc
        raise

Esse padrão ajuda quem chama a função. A camada de negócio não precisa conhecer todos os detalhes de HTTPX; ela decide o que fazer quando o CRM está indisponível ou quando o cliente não existe.

Retries: quando tentar de novo

Retry não é martelo universal. Tentar de novo uma operação errada só aumenta carga e pode duplicar efeitos colaterais. A regra prática:

HTTPX não traz uma política de retry completa embutida para todos os cenários. Você pode usar bibliotecas como tenacity, configurar transporte customizado ou escrever um retry pequeno quando a regra for simples. Para muitos projetos, um loop explícito é mais fácil de auditar:

import time
import httpx

STATUS_TEMPORARIOS = {429, 500, 502, 503, 504}


def get_com_retry(client: httpx.Client, url: str, tentativas: int = 3) -> httpx.Response:
    ultimo_erro: Exception | None = None

    for tentativa in range(1, tentativas + 1):
        try:
            response = client.get(url)
            if response.status_code not in STATUS_TEMPORARIOS:
                response.raise_for_status()
                return response

            if tentativa == tentativas:
                response.raise_for_status()

            espera = calcular_espera(response, tentativa)
        except (httpx.TimeoutException, httpx.NetworkError) as exc:
            ultimo_erro = exc
            if tentativa == tentativas:
                raise
            espera = min(2 ** tentativa, 10)

        time.sleep(espera)

    raise RuntimeError("Retry esgotado") from ultimo_erro


def calcular_espera(response: httpx.Response, tentativa: int) -> float:
    retry_after = response.headers.get("Retry-After")
    if retry_after and retry_after.isdigit():
        return min(float(retry_after), 60.0)
    return min(2 ** tentativa, 10.0)

O detalhe importante é limitar tentativas e espera. Sem limite, retry vira loop infinito. Em sistemas com muitos workers, retries simultâneos também podem virar uma tempestade contra a API. Quando o volume cresce, combine retry com fila, rate limit e observabilidade.

Assíncrono com AsyncClient

HTTPX também funciona bem em código assíncrono. Isso é útil em FastAPI, crawlers responsáveis, bots e integrações que precisam esperar várias APIs sem bloquear a thread principal.

import httpx

async def buscar_cotacao(moeda: str) -> dict:
    timeout = httpx.Timeout(connect=2.0, read=5.0, write=5.0, pool=2.0)

    async with httpx.AsyncClient(
        base_url="https://api.example.com",
        timeout=timeout,
    ) as client:
        response = await client.get(f"/cotacoes/{moeda}")
        response.raise_for_status()
        return response.json()

Não misture cliente síncrono dentro de endpoint assíncrono se a chamada pode demorar. Isso bloqueia o event loop e reduz a capacidade da aplicação. Em FastAPI, use AsyncClient para chamadas externas em endpoints async def, ou mova trabalho pesado para fila quando a operação não precisa responder imediatamente.

Limites de conexão e backpressure

Timeout define quanto esperar. Limite de conexão define quantas chamadas simultâneas o cliente permite. Sem limite, uma explosão de tráfego pode abrir conexões demais, piorar a instabilidade e causar bloqueio em cascata.

limits = httpx.Limits(
    max_connections=20,
    max_keepalive_connections=10,
    keepalive_expiry=30.0,
)

client = httpx.Client(
    base_url="https://api.example.com",
    timeout=httpx.Timeout(connect=2.0, read=5.0, write=5.0, pool=2.0),
    limits=limits,
)

Pense nesses números como contrato com a API externa e com sua própria aplicação. Um job noturno pode usar concorrência maior. Um endpoint público que chama serviço de terceiros deve ser mais conservador, porque cada requisição de usuário vira uma requisição externa.

Logging sem vazar dados sensíveis

Quando uma API falha, você precisa saber qual endpoint, status, tempo e correlação do pedido. Mas logs não podem expor token, CPF, email, cartão, segredo ou payload sensível.

Um formato seguro registra metadados:

import logging
import time

logger = logging.getLogger(__name__)


def consultar_pedido(client, pedido_id: str) -> dict:
    inicio = time.perf_counter()
    try:
        response = client.get(f"/pedidos/{pedido_id}")
        response.raise_for_status()
        return response.json()
    finally:
        duracao_ms = int((time.perf_counter() - inicio) * 1000)
        logger.info(
            "crm_request",
            extra={
                "endpoint": "/pedidos/{id}",
                "duracao_ms": duracao_ms,
                "pedido_id_hash": hash(pedido_id),
            },
        )

Em sistemas maiores, prefira IDs de correlação e métricas agregadas: taxa de erro por status, latência p95, quantidade de retries e chamadas bloqueadas por rate limit. Isso conversa diretamente com observabilidade em Python com OpenTelemetry.

Testando clientes HTTPX

Não teste integração externa batendo na API real em toda execução. Isso deixa a suíte lenta, cara e instável. O ideal é testar sua lógica com respostas controladas e reservar poucos testes manuais ou de contrato para a API real.

HTTPX permite usar MockTransport:

import httpx


def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/clientes/123":
        return httpx.Response(200, json={"id": "123", "nome": "Ana"})
    return httpx.Response(404, json={"erro": "não encontrado"})


def test_buscar_cliente():
    transport = httpx.MockTransport(handler)
    client = httpx.Client(base_url="https://crm.test", transport=transport)

    assert buscar_cliente(client, "123") == {"id": "123", "nome": "Ana"}

Esse teste roda rápido e cobre sua regra de tratamento. Para timeouts, você pode fazer o handler levantar httpx.ReadTimeout. Para 429, devolva header Retry-After e valide se sua função respeita a política definida.

Checklist para produção

Antes de colocar um cliente HTTP em produção, revise:

1. Todas as chamadas têm timeout explícito.
2. O código usa Client ou AsyncClient reaproveitado.
3. Erros de rede, timeout e status HTTP são tratados separadamente.
4. Retries são limitados, com backoff e só para operações seguras.
5. 429 respeita Retry-After quando disponível.
6. Operações com efeito colateral usam chave de idempotência quando a API suporta.
7. Logs não expõem tokens, documentos, emails ou payload sensível.
8. Testes usam mock/transport controlado para cenários previsíveis.
9. Métricas acompanham latência, taxa de erro e retries.
10. A aplicação tem plano para degradação: fila, cache, resposta parcial ou mensagem clara ao usuário.

Exemplo de estrutura de projeto

Para uma integração pequena, esta organização já ajuda:

integracao-crm/
  pyproject.toml
  src/
    integracao_crm/
      __init__.py
      client.py
      errors.py
      models.py
      settings.py
  tests/
    test_client.py
    test_retries.py

client.py concentra HTTPX. errors.py traduz falhas para exceções do domínio. settings.py lê URL, token e timeouts via variável de ambiente. Os testes validam comportamento sem depender da internet. É uma base simples, mas muito mais profissional do que espalhar httpx.get() pelo código inteiro.

Conclusão

HTTPX facilita escrever clientes HTTP modernos em Python, mas confiabilidade vem das escolhas em volta da biblioteca. Timeout explícito evita travamento. Client reaproveitado reduz overhead. Tratamento de erro deixa a camada de negócio mais clara. Retry limitado protege contra falhas temporárias sem multiplicar problemas. Testes com MockTransport tornam a integração previsível.

Para quem está montando portfólio ou trabalhando em produto real, esse tipo de cuidado diferencia um script que “funciona na minha máquina” de uma automação que aguenta produção. Depois deste guia, vale aprofundar em background tasks, Celery e Redis para tarefas demoradas e em feature flags com Python para reduzir risco ao lançar integrações novas.

Se o objetivo for comparar esse padrão com outras stacks, veja também como a comunidade irmã aborda APIs e backends em Go Brasil, Rust Brasil e Kotlin Brasil. Cada ecossistema tem ferramentas diferentes, mas timeouts, retries limitados e observabilidade continuam sendo princípios comuns.