Chapter 51, Advanced
Architecture and errors in production
Code that works on your machine is the beginning. Production requires it to fail predictably, be testable without infrastructure, and show what is happening at three in the morning.
Code for this chapter: avancado/cap51_arquitetura.py
Separate business logic from the outside world
The architectural decision that yields the most return: the domain logic does not care about database, network, or clock. It receives interfaces (which we describe with Protocol) and whoever assembles it decides the implementations. In testing, in-memory versions come in. In production, the real ones:
from dataclasses import dataclass
from typing import Protocol
@dataclass(frozen=True)
class Pedido:
id: str
cliente: str
total: float
class RepositorioPedidos(Protocol):
def salvar(self, pedido: Pedido) -> None: ...
def existe(self, pedido_id: str) -> bool: ...
class Notificador(Protocol):
def enviar(self, destinatario: str, mensagem: str) -> None: ...
class PedidoDuplicado(Exception):
pass
class ServicoPedidos:
def __init__(self, repositorio: RepositorioPedidos, notificador: Notificador) -> None:
self._repositorio = repositorio
self._notificador = notificador
def registrar(self, pedido: Pedido) -> None:
if self._repositorio.existe(pedido.id):
raise PedidoDuplicado(pedido.id)
self._repositorio.salvar(pedido)
self._notificador.enviar(pedido.cliente, f"pedido {pedido.id} recebido")
The service does not know if the repository is PostgreSQL or a dictionary. Therefore, testing it does not require spinning up anything:
class RepositorioEmMemoria:
def __init__(self) -> None:
self._dados: dict[str, Pedido] = {}
def salvar(self, pedido: Pedido) -> None:
self._dados[pedido.id] = pedido
def existe(self, pedido_id: str) -> bool:
return pedido_id in self._dados
class NotificadorFalso:
def __init__(self) -> None:
self.enviadas: list[tuple[str, str]] = []
def enviar(self, destinatario: str, mensagem: str) -> None:
self.enviadas.append((destinatario, mensagem))
notificador = NotificadorFalso()
servico = ServicoPedidos(RepositorioEmMemoria(), notificador)
servico.registrar(Pedido("p1", "ana", 50.0))
print(notificador.enviadas)
try:
servico.registrar(Pedido("p1", "ana", 50.0))
except PedidoDuplicado as erro:
print("duplicado:", erro)
[('ana', 'pedido p1 recebido')]
duplicado: p1
Checking and then writing is not atomic
The
existefollowed bysalvarhas a race window: two identical requests, in parallel, pass through the check before either writes. In production, the guarantee of uniqueness comes from the database (a restrictionUNIQUE), and the service handles the error it returns. The check in the code is just a courtesy.
Configuration outside of code
The configuration comes from the environment (variables), and is never not from the code or the repository. I read everything in one place, at startup, and fail early if something mandatory is missing. A program that crashes at startup with a clear message is much better than one that crashes an hour later with an obscure error:
import os
from collections.abc import Mapping
from dataclasses import dataclass
@dataclass(frozen=True)
class Configuracao:
ambiente: str
timeout_segundos: float
url_banco: str
@classmethod
def do_ambiente(cls, env: Mapping[str, str] | None = None) -> "Configuracao":
env = os.environ if env is None else env
return cls(
ambiente=env.get("APP_AMBIENTE", "desenvolvimento"),
timeout_segundos=float(env.get("APP_TIMEOUT", "5")),
url_banco=env["APP_URL_BANCO"],
)
config = Configuracao.do_ambiente(
{"APP_URL_BANCO": "postgresql://localhost/app", "APP_TIMEOUT": "2.5"}
)
print(config)
try:
Configuracao.do_ambiente({})
except KeyError as erro:
print("variável obrigatória ausente:", erro)
Configuracao(ambiente='desenvolvimento', timeout_segundos=2.5, url_banco='postgresql://localhost/app')
variável obrigatória ausente: 'APP_URL_BANCO'
Secrets (passwords, API keys) come from a secrets manager or from the runtime environment variables. They never enter Git, nor logs.
Retries with exponential backoff
Transient failures (network, brief unavailability) deserve a retry, with a wait that grows each time (exponential backoff) and a random variation (jitter), so that a thousand clients do not return all at the same moment. I inject the wait function and the random function as parameters, and this makes the behavior testable without actually waiting:
import random
import time
from collections.abc import Callable
def com_retentativas[T](
operacao: Callable[[], T],
*,
tentativas: int = 4,
base: float = 0.5,
dormir: Callable[[float], None] = time.sleep,
jitter: Callable[[], float] = random.random,
) -> T:
for numero in range(1, tentativas + 1):
try:
return operacao()
except ConnectionError:
if numero == tentativas:
raise
espera = base * 2 ** (numero - 1) * (0.5 + jitter() / 2)
dormir(espera)
raise AssertionError("inalcançável")
esperas: list[float] = []
chamadas = {"total": 0}
def instavel() -> str:
chamadas["total"] += 1
if chamadas["total"] < 3:
raise ConnectionError("falha")
return "ok"
resultado = com_retentativas(instavel, dormir=esperas.append, jitter=lambda: 1.0)
print(resultado, esperas)
ok [0.5, 1.0]
Three rules that accompany retries: every external call has a timeout (without a timeout, a slow dependency blocks your service); only repeat idempotent operations (repeating "consult" is safe, repeating "charge" is not, unless there is an idempotency key); and set a limit on attempts, so that the failure does not turn into an avalanche.
Logs that connect
When a thousand requests run at the same time, the log lines get mixed up. A correlation identifier in each line allows reconstructing the path of a request. The contextvars securely stores this value in both threads and asyncio:
import contextvars
import json
import logging
import sys
id_requisicao = contextvars.ContextVar("id_requisicao", default="-")
class FiltroDeContexto(logging.Filter):
def filter(self, record: logging.LogRecord) -> bool:
setattr(record, "id_requisicao", id_requisicao.get())
return True
class FormatoJson(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
return json.dumps(
{
"nivel": record.levelname,
"mensagem": record.getMessage(),
"id_requisicao": getattr(record, "id_requisicao", "-"),
},
ensure_ascii=False,
)
manipulador = logging.StreamHandler(sys.stdout)
manipulador.setFormatter(FormatoJson())
manipulador.addFilter(FiltroDeContexto())
log = logging.getLogger("api")
log.addHandler(manipulador)
log.setLevel(logging.INFO)
log.propagate = False
def tratar(id_: str) -> None:
token = id_requisicao.set(id_)
try:
log.info("pedido processado")
finally:
id_requisicao.reset(token)
tratar("req-1")
tratar("req-2")
log.info("fora de requisição")
{"nivel": "INFO", "mensagem": "pedido processado", "id_requisicao": "req-1"}
{"nivel": "INFO", "mensagem": "pedido processado", "id_requisicao": "req-2"}
{"nivel": "INFO", "mensagem": "fora de requisição", "id_requisicao": "-"}
Errors: translate at the boundary
Within the system, use domain exceptions (PedidoDuplicado). At the boundary (the API, the CLI), translate to what the outside world understands and is never leak internal details in the response: the traceback goes to the log, and the caller receives a safe message:
def para_resposta(excecao: Exception) -> tuple[int, str]:
if isinstance(excecao, PedidoDuplicado):
return 409, "pedido já registrado"
if isinstance(excecao, ValueError):
return 422, str(excecao)
return 500, "erro interno"
print(para_resposta(PedidoDuplicado("p1")), para_resposta(RuntimeError("segredo")))
(409, 'pedido já registrado') (500, 'erro interno')
Questions I ask before approving a service
- What happens when each dependency becomes slow? And when it goes down?
- Every external call has timeout and explicit retry policy?
- The operation can be repeated without side effects (is it idempotent)?
- Can you reproduce a request from the logs, with an identifier?
- The configuration comes from the environment, and the absence of a mandatory value breaks the startup?
- What is the plan to rollback the deploy, and what happens to the data already written?
- What breaks first with ten times more load?
Exercise 1
Test the duplication rule
Write a test for ServicoPedidos that ensures that a duplicate order raises PedidoDuplicado and does not sends the second notification.