Chapter 52, Backend
HTTP and API consumption
Almost every real system communicates with another via HTTP. Learning to consume an API with timeout, error handling, and retry is what separates a script from a reliable integration.
Code for this chapter: backend/cap52_http_apis.py
What happens in a request
An HTTP request is a text with four parts: the method (what to do), the URL (where), the headers (metadata like format and credentials) and, sometimes, a body. The response has a status code, headers, and a body. Almost everything you will debug is in one of these parts.
| Status range | Meaning | What the client does |
|---|---|---|
2xx | It worked | Move on |
3xx | Redirection | The HTTP client usually follows on its own |
4xx | Client error (you requested incorrectly) | No repeat the same: correct the request |
5xx | Server error | It may be worth a new attempt |
The distinction between 4xx and 5xx decides your error strategy. Repeating a 404 a thousand times does not turn it into 200. Repeating a 503 many times may work.
A local server to practice
To ensure the examples have predictable output, without relying on the internet, I set up a mock server on my own machine, using only the standard library. It has a stable endpoint, an unstable one (fails twice and then works), a slow one, a paginated one, and one that receives JSON. You don't need to understand its code, just know that the routes exist:
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]}"
A request with httpx2
The httpx2 is the HTTP client I recommend today. It has the same simple interface as requests (which is still widely used), and adds asynchronous client, timeouts by default, and support for HTTP/2 (installed with httpx2[http2]). Always use a Client within a with: it reuses the connection between calls, which is much faster than opening a new one for each request:
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())
200 application/json {'mensagem': 'olá'}
HTTP error is not an exception
An 404 or a 503 does not raise exceptions on their own: the response arrives normally, and you need to check the status. The method raise_for_status() converts any 4xx or 5xx into an exception, which is usually what you want:
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)
404 False
erro HTTP: 404
Timeout: always
A network call without a timeout can wait forever and freeze your entire service. Every external call needs an explicit time limit:
try:
httpx2.get(f"{BASE}/lento", timeout=0.2)
except httpx2.TimeoutException as erro:
print("estourou o tempo:", type(erro).__name__)
estourou o tempo: ReadTimeout
Retries with exponential backoff
Failures 5xx and network issues tend to be transient. The reasonable policy is to try again a few times, waiting longer with each attempt (exponential backoff), and give up. As in the architecture chapter, I inject the wait function to be able to test without actually waiting:
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)
200 {'tentativa': 3} [0.1, 0.2]
Just repeat what is safe to repeat
Repeating a
GETis safe. Repeating aPOSTthat creates something may create it twice. Only do automatic retries on idempotent, or send an idempotency key (next chapter).
Pagination
APIs do not return millions of items at once. They deliver in pages, and the client continues until it’s finished. A generator encapsulates this, and the consumer sees a continuous flow:
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)))
[0, 1, 2, 3, 4, 5, 6, 7, 8]
Send JSON
The argument json= serializes the dictionary and already sets the header Content-Type: application/json:
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())
201 {'recebido': {'cliente': 'Ana', 'itens': 2}}
Multiple requests at the same time
To trigger many independent calls, the AsyncClient with asyncio.gather (chapter 47) works in a single thread:
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()))
[200, 200, 200]
httpx and httpx2
The
httpxclassic (version 0.28) have the same interface, and you can find it in a lot of existing code. Thehttpx2presents itself as the next generation of the same client, and is maintained by Pydantic. Starlette, the foundation of FastAPI, has already started to prefer thehttpx2in itsTestClientand issues a deprecation warning if it only finds thehttpx. For new code, I use thehttpx2. For everything this chapter uses, the migration consists of changing the package name and theimport: I ran all the examples with thehttpx2, without any other changes.
What I do in every integration
Reused client, explicit timeout,
raise_for_status(or deliberate status handling), retry only for what is safe, credentials read from the environment (never in the code) and no secrets in the logs.
Exercise 1
A client that translates HTTP error into domain exception
Write buscar_json(cliente, caminho) that returns JSON on success and raises ErroDeApi (with the attribute status) for any response that is not 2xx.