Pular para o conteúdo

    Capítulo 52, Backend

    HTTP e consumo de APIs

    Quase todo sistema real conversa com outro por HTTP. Aprender a consumir uma API com timeout, tratamento de erro e retentativa é o que separa um script de uma integração confiável.

    Código deste capítulo: backend/cap52_http_apis.py

    O que acontece em uma requisição

    Uma requisição HTTP é um texto com quatro partes: o método (o que fazer), a URL (onde), os cabeçalhos (metadados como formato e credenciais) e, às vezes, um corpo. A resposta tem um código de status, cabeçalhos e um corpo. Quase tudo que você vai depurar está em uma dessas partes.

    Faixa do statusSignificadoO que o cliente faz
    2xxDeu certoSegue em frente
    3xxRedirecionamentoO cliente HTTP normalmente segue sozinho
    4xxErro do cliente (você pediu errado)Não repita igual: corrija o pedido
    5xxErro do servidorPode valer uma nova tentativa

    A distinção entre 4xx e 5xx decide a sua estratégia de erro. Repetir um 404 mil vezes não o transforma em 200. Repetir um 503 muitas vezes pode funcionar.

    Um servidor local para praticar

    Para os exemplos terem saída previsível, sem depender de internet, eu subo um servidor de brincadeira na própria máquina, usando só a biblioteca padrão. Ele tem um endpoint estável, um instável (falha duas vezes e depois funciona), um lento, um paginado e um que recebe JSON. Você não precisa entender o código dele, apenas saber que as rotas existem:

    backend/cap52_http_apis.pylinhas 10 a 61
    import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

import httpx2

contador_instavel = {"n": 0}


class Manipulador(BaseHTTPRequestHandler):
    def responder(self, status, corpo):
        dados = json.dumps(corpo).encode()
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(dados)))
        self.end_headers()
        self.wfile.write(dados)

    def do_GET(self):
        url = urlparse(self.path)
        if url.path == "/ok":
            self.responder(200, {"mensagem": "olá"})
        elif url.path == "/instavel":
            contador_instavel["n"] += 1
            if contador_instavel["n"] < 3:
                self.responder(503, {"erro": "indisponível"})
            else:
                self.responder(200, {"tentativa": contador_instavel["n"]})
        elif url.path == "/lento":
            time.sleep(1)
            self.responder(200, {})
        elif url.path == "/itens":
            pagina = int(parse_qs(url.query).get("pagina", ["1"])[0])
            proxima = pagina + 1 if pagina < 3 else None
            self.responder(200, {"itens": list(range((pagina - 1) * 3, pagina * 3)), "proxima": proxima})
        else:
            self.responder(404, {"erro": "não encontrado"})

    def do_POST(self):
        tamanho = int(self.headers.get("Content-Length", 0))
        corpo = json.loads(self.rfile.read(tamanho) or b"{}")
        self.responder(201, {"recebido": corpo})

    def log_message(self, *args):
        pass


servidor = ThreadingHTTPServer(("127.0.0.1", 0), Manipulador)
threading.Thread(target=servidor.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{servidor.server_address[1]}"

    Uma requisição com httpx2

    O httpx2 é o cliente HTTP que eu recomendo hoje. Ele tem a mesma interface simples do requests (que continua muito usado), e acrescenta cliente assíncrono, timeouts por padrão e suporte a HTTP/2 (instalado com httpx2[http2]). Use sempre um Client dentro de um with: ele reaproveita a conexão entre as chamadas, o que é muito mais rápido do que abrir uma nova a cada requisição:

    backend/cap52_http_apis.pylinhas 66 a 68
    with httpx2.Client(base_url=BASE, timeout=2.0) as cliente:
    resposta = cliente.get("/ok")
    print(resposta.status_code, resposta.headers["content-type"], resposta.json())
    Saída
    200 application/json {'mensagem': 'olá'}

    Erro HTTP não é exceção

    Um 404 ou um 503 não levantam exceção sozinhos: a resposta chega normalmente, e você precisa olhar o status. O método raise_for_status() converte qualquer 4xx ou 5xx em exceção, o que costuma ser o que você quer:

    backend/cap52_http_apis.pylinhas 73 a 79
    with httpx2.Client(base_url=BASE, timeout=2.0) as cliente:
    resposta = cliente.get("/nao-existe")
    print(resposta.status_code, resposta.is_success)
    try:
        resposta.raise_for_status()
    except httpx2.HTTPStatusError as erro:
        print("erro HTTP:", erro.response.status_code)
    Saída
    404 False
erro HTTP: 404

    Timeout: sempre

    Uma chamada de rede sem timeout pode ficar esperando para sempre e travar o seu serviço inteiro. Toda chamada externa precisa de um limite de tempo explícito:

    backend/cap52_http_apis.pylinhas 84 a 87
    try:
    httpx2.get(f"{BASE}/lento", timeout=0.2)
except httpx2.TimeoutException as erro:
    print("estourou o tempo:", type(erro).__name__)
    Saída
    estourou o tempo: ReadTimeout

    Retentativas com espera crescente

    Falhas 5xx e de rede costumam ser transitórias. A política razoável é tentar de novo poucas vezes, esperando mais a cada tentativa (backoff exponencial), e desistir. Como no capítulo de arquitetura, eu injeto a função de espera para poder testar sem esperar de verdade:

    backend/cap52_http_apis.pylinhas 92 a 110
    def get_com_retentativas(cliente, caminho, *, tentativas=4, base=0.1, dormir=time.sleep):
    ultima = None
    for numero in range(1, tentativas + 1):
        try:
            ultima = cliente.get(caminho)
            if ultima.status_code < 500:
                return ultima
        except httpx2.TransportError:
            if numero == tentativas:
                raise
        if numero < tentativas:
            dormir(base * 2 ** (numero - 1))
    return ultima


esperas = []
with httpx2.Client(base_url=BASE, timeout=2.0) as cliente:
    resposta = get_com_retentativas(cliente, "/instavel", dormir=esperas.append)
print(resposta.status_code, resposta.json(), esperas)
    Saída
    200 {'tentativa': 3} [0.1, 0.2]

    Só repita o que é seguro repetir

    Repetir um GET é seguro. Repetir um POST que cria algo pode criar duas vezes. Só faça retentativa automática em operações idempotentes, ou envie uma chave de idempotência (capítulo seguinte).

    Paginação

    APIs não devolvem milhões de itens de uma vez. Elas entregam por páginas, e o cliente segue até acabar. Um gerador encapsula isso e quem consome vê um fluxo contínuo:

    backend/cap52_http_apis.pylinhas 115 a 124
    def todos_os_itens(cliente):
    pagina = 1
    while pagina is not None:
        dados = cliente.get("/itens", params={"pagina": pagina}).json()
        yield from dados["itens"]
        pagina = dados["proxima"]


with httpx2.Client(base_url=BASE, timeout=2.0) as cliente:
    print(list(todos_os_itens(cliente)))
    Saída
    [0, 1, 2, 3, 4, 5, 6, 7, 8]

    Enviar JSON

    O argumento json= serializa o dicionário e já define o cabeçalho Content-Type: application/json:

    backend/cap52_http_apis.pylinhas 129 a 131
    with httpx2.Client(base_url=BASE, timeout=2.0) as cliente:
    resposta = cliente.post("/pedidos", json={"cliente": "Ana", "itens": 2})
    print(resposta.status_code, resposta.json())
    Saída
    201 {'recebido': {'cliente': 'Ana', 'itens': 2}}

    Várias requisições ao mesmo tempo

    Para disparar muitas chamadas independentes, o AsyncClient com asyncio.gather (capítulo 47) trabalha em uma única thread:

    backend/cap52_http_apis.pylinhas 136 a 145
    import asyncio


async def buscar_varios():
    async with httpx2.AsyncClient(base_url=BASE, timeout=2.0) as cliente:
        respostas = await asyncio.gather(*(cliente.get("/ok") for _ in range(3)))
    return [r.status_code for r in respostas]


print(asyncio.run(buscar_varios()))
    Saída
    [200, 200, 200]

    O httpx e o httpx2

    O httpx clássico (versão 0.28) tem a mesma interface, e você o encontra em muito código existente. O httpx2 se apresenta como a próxima geração do mesmo cliente, e é mantido pela Pydantic. O Starlette, a base do FastAPI, já passou a preferir o httpx2 no seu TestClient e emite um aviso de depreciação se encontrar só o httpx. Para código novo, eu uso o httpx2. Para tudo o que este capítulo usa, a migração consiste em trocar o nome do pacote e do import: eu executei todos os exemplos com o httpx2, sem nenhuma outra mudança.

    O que eu faço em toda integração

    Cliente reutilizado, timeout explícito, raise_for_status (ou tratamento deliberado do status), retentativa só para o que é seguro, credenciais lidas do ambiente (nunca no código) e nenhum segredo nos logs.

    Exercício 1

    Um cliente que traduz erro HTTP em exceção do domínio

    Escreva buscar_json(cliente, caminho) que devolva o JSON em caso de sucesso e levante ErroDeApi (com o atributo status) para qualquer resposta que não seja 2xx.