Pular para o conteúdo

    Capítulo 51, Avançado

    Arquitetura e erros em produção

    Código que funciona na sua máquina é o começo. Produção exige que ele falhe de forma previsível, seja testável sem infraestrutura e mostre o que está acontecendo às três da manhã.

    Código deste capítulo: avancado/cap51_arquitetura.py

    Separar a regra de negócio do mundo externo

    A decisão de arquitetura que mais retorno dá: a lógica do domínio não importa banco, rede nem relógio. Ela recebe interfaces (que descrevemos com Protocol) e quem a monta decide as implementações. Em teste, entram versões em memória. Em produção, as reais:

    avancado/cap51_arquitetura.pylinhas 10 a 43
    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")

    O serviço não sabe se o repositório é PostgreSQL ou um dicionário. Por isso, testá-lo não exige subir nada:

    avancado/cap51_arquitetura.pylinhas 45 a 71
    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)
    Saída
    [('ana', 'pedido p1 recebido')]
duplicado: p1

    Verificar e depois gravar não é atômico

    O existe seguido de salvar tem uma janela de corrida: dois pedidos iguais, em paralelo, passam pela verificação antes de qualquer um gravar. Em produção, a garantia de unicidade vem do banco (uma restrição UNIQUE), e o serviço trata o erro que ela devolve. A verificação no código é só uma cortesia.

    Configuração fora do código

    A configuração vem do ambiente (variáveis), e nunca do código nem do repositório. Eu leio tudo em um único lugar, na inicialização, e falho cedo se faltar algo obrigatório. O programa que cai na partida com uma mensagem clara é muito melhor do que o que cai uma hora depois com um erro obscuro:

    avancado/cap51_arquitetura.pylinhas 76 a 104
    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)
    Saída
    Configuracao(ambiente='desenvolvimento', timeout_segundos=2.5, url_banco='postgresql://localhost/app')
variável obrigatória ausente: 'APP_URL_BANCO'

    Segredos (senhas, chaves de API) vêm de um gerenciador de segredos ou das variáveis do ambiente de execução. Eles nunca entram no Git, nem em logs.

    Retentativas com espera crescente

    Falhas transitórias (rede, indisponibilidade breve) merecem uma nova tentativa, com espera que cresce a cada vez (backoff exponencial) e um sorteio de variação (jitter), para que mil clientes não voltem todos no mesmo instante. Eu injeto a função de espera e a de sorteio como parâmetros, e isso torna o comportamento testável sem esperar de verdade:

    avancado/cap51_arquitetura.pylinhas 109 a 145
    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)
    Saída
    ok [0.5, 1.0]

    Três regras que acompanham as retentativas: toda chamada externa tem um tempo limite (sem timeout, uma dependência lenta trava o seu serviço); só repita operações idempotentes (repetir "consultar" é seguro, repetir "cobrar" não é, a menos que exista uma chave de idempotência); e ponha um limite de tentativas, para a falha não virar uma avalanche.

    Logs que se conectam

    Quando mil requisições rodam ao mesmo tempo, as linhas de log se misturam. Um identificador de correlação em cada linha permite reconstruir o caminho de uma requisição. O contextvars guarda esse valor de forma segura tanto em threads quanto em asyncio:

    avancado/cap51_arquitetura.pylinhas 150 a 195
    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")
    Saída
    {"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": "-"}

    Erros: traduza na fronteira

    Dentro do sistema, use exceções do domínio (PedidoDuplicado). Na fronteira (a API, a CLI), traduza para o que o mundo externo entende e nunca vaze detalhes internos na resposta: o traceback vai para o log, e quem chamou recebe uma mensagem segura:

    avancado/cap51_arquitetura.pylinhas 200 a 208
    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")))
    Saída
    (409, 'pedido já registrado') (500, 'erro interno')

    Perguntas que eu faço antes de aprovar um serviço

    • O que acontece quando cada dependência fica lenta? E quando fica fora do ar?
    • Toda chamada externa tem timeout e política de retentativa explícita?
    • A operação pode ser repetida sem efeito colateral (é idempotente)?
    • Dá para reproduzir uma requisição a partir dos logs, com um identificador?
    • A configuração vem do ambiente, e a ausência de um valor obrigatório derruba a partida?
    • Qual é o plano de reverter o deploy, e o que acontece com os dados já gravados?
    • O que quebra primeiro com dez vezes mais carga?

    Exercício 1

    Teste a regra de duplicidade

    Escreva um teste para ServicoPedidos que garanta que um pedido duplicado levanta PedidoDuplicado e não envia a segunda notificação.