/
Capítulo 62, Backend
API completa: do contrato ao deploy
Este capítulo junta tudo o que os anteriores ensinaram em uma API de pedidos que funciona de ponta a ponta. Cada arquivo abaixo foi executado: os testes passam contra SQLite e contra PostgreSQL, e o mypy estrito e o ruff aprovam o código.
Os arquivos deste capítulo estão em exemplos/api_pedidos/.
O que a API faz
É uma API de pedidos com cinco rotas, pensada para mostrar decisões de produção, e não só rotas que funcionam:
| Rota | O que faz | Detalhe de produção |
|---|---|---|
POST /pedidos | Cria um pedido | Idempotency-Key evita pedido duplicado |
GET /pedidos/{id} | Lê um pedido | Erro 404 no formato problem+json |
GET /pedidos | Lista com paginação | Cursor, e não deslocamento |
POST /pedidos/{id}/cancelar | Cancela | 409 se já estiver cancelado |
GET /saude e GET /metricas | Saúde e métricas | Para o orquestrador e o Prometheus |
A estrutura em camadas
src/api_pedidos/
app.py HTTP: rotas, status, erros (conhece FastAPI)
servico.py Regras de negócio (não conhece HTTP)
repositorio.py Consultas ao banco (conhece SQLAlchemy)
modelos.py Tabelas (SQLAlchemy)
esquemas.py Contrato de entrada e saída (Pydantic)
config.py Configuração do ambiente (pydantic-settings)
db.py Engine e sessão
observabilidade.py Logs JSON, id de requisição, métricas
migrations/ Histórico do esquema (Alembic)
tests/ Testes de API, de serviço e de migração
A seta de dependência vai sempre de cima para baixo: a rota chama o serviço, o serviço chama o repositório. O serviço não importa o FastAPI, e por isso dá para testá-lo sem HTTP e reaproveitá-lo em uma CLI ou em um worker.
Dependências e ferramentas
O pyproject.toml declara as dependências de produção, as de desenvolvimento (em um grupo separado, que o Dockerfile não instala) e a configuração do pytest, do ruff e do mypy:
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "api-pedidos"
version = "0.1.0"
description = "API de pedidos do livro Python na Prática"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"fastapi>=0.115",
"uvicorn>=0.30",
"sqlalchemy>=2.0",
"alembic>=1.13",
"psycopg[binary]>=3.2",
"pydantic-settings>=2.4",
"prometheus-client>=0.20",
]
[dependency-groups]
dev = [
"pytest>=8",
"httpx2>=2.0",
"mypy>=1.10",
"ruff>=0.6",
]
[tool.pytest.ini_options]
testpaths = ["tests"]
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP"]
[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]
files = ["src"]
Configuração e banco
A configuração é a do capítulo 58, e a guarda prod com SQLite impede o erro mais comum de deploy:
from functools import lru_cache
from typing import Literal, Self
from pydantic import SecretStr, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Configuracao(BaseSettings):
model_config = SettingsConfigDict(env_prefix="API_", env_file=".env", extra="ignore")
ambiente: Literal["dev", "test", "prod"] = "dev"
database_url: SecretStr = SecretStr("sqlite:///./dev.db")
log_nivel: str = "INFO"
@model_validator(mode="after")
def exigir_banco_de_servidor_em_producao(self) -> Self:
if self.ambiente == "prod" and self.database_url.get_secret_value().startswith("sqlite"):
raise ValueError("produção exige um banco de dados de servidor, não SQLite")
return self
@lru_cache
def obter_configuracao() -> Configuracao:
return Configuracao()
O modelo declara as restrições do banco: chave única de idempotência, índice por cliente e chave estrangeira com ON DELETE CASCADE. O dinheiro fica em centavos inteiros:
from datetime import UTC, datetime
from sqlalchemy import DateTime, ForeignKey, Index, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
def agora() -> datetime:
return datetime.now(UTC)
class Pedido(Base):
__tablename__ = "pedidos"
__table_args__ = (Index("ix_pedidos_cliente", "cliente"),)
id: Mapped[int] = mapped_column(primary_key=True)
cliente: Mapped[str] = mapped_column(String(120))
status: Mapped[str] = mapped_column(String(20), default="aberto")
chave_idempotencia: Mapped[str | None] = mapped_column(String(80), unique=True)
criado_em: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=agora)
itens: Mapped[list["ItemPedido"]] = relationship(
back_populates="pedido", cascade="all, delete-orphan"
)
@property
def total_centavos(self) -> int:
return sum(item.quantidade * item.preco_centavos for item in self.itens)
class ItemPedido(Base):
__tablename__ = "itens_pedido"
id: Mapped[int] = mapped_column(primary_key=True)
pedido_id: Mapped[int] = mapped_column(ForeignKey("pedidos.id", ondelete="CASCADE"))
produto: Mapped[str] = mapped_column(String(120))
quantidade: Mapped[int]
preco_centavos: Mapped[int]
pedido: Mapped[Pedido] = relationship(back_populates="itens")
from collections.abc import Iterator
from typing import Any
from fastapi import Request
from sqlalchemy import Engine, create_engine
from sqlalchemy.orm import Session, sessionmaker
def criar_engine(url: str) -> Engine:
argumentos: dict[str, Any] = {}
if url.startswith("sqlite"):
argumentos["check_same_thread"] = False
return create_engine(url, connect_args=argumentos, pool_pre_ping=True)
def criar_fabrica_de_sessoes(engine: Engine) -> sessionmaker[Session]:
return sessionmaker(engine, expire_on_commit=False)
def obter_sessao(request: Request) -> Iterator[Session]:
with request.app.state.fabrica() as sessao:
yield sessao
O contrato
Os esquemas Pydantic são o contrato: validam a entrada (quantidade positiva, no mínimo um item) e definem exatamente o que a saída contém. O FastAPI deriva a documentação OpenAPI deles:
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class ItemCriar(BaseModel):
produto: str = Field(min_length=1, max_length=120)
quantidade: int = Field(gt=0, le=1000)
preco_centavos: int = Field(gt=0)
class PedidoCriar(BaseModel):
cliente: str = Field(min_length=1, max_length=120)
itens: list[ItemCriar] = Field(min_length=1)
class ItemLer(ItemCriar):
model_config = ConfigDict(from_attributes=True)
class PedidoLer(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
cliente: str
status: Literal["aberto", "cancelado"]
total_centavos: int
criado_em: datetime
itens: list[ItemLer]
class Pagina(BaseModel):
itens: list[PedidoLer]
proximo_cursor: int | None
Repositório e serviço
O repositório só sabe consultar. O serviço decide. Repare em criar: a verificação da chave e o tratamento do IntegrityError coexistem, porque a verificação evita o trabalho na maioria das vezes e a restrição UNIQUE do banco garante a correção quando duas requisições disputam a mesma chave:
from sqlalchemy import select
from sqlalchemy.orm import Session
from api_pedidos.modelos import Pedido
class RepositorioPedidos:
def __init__(self, sessao: Session) -> None:
self._sessao = sessao
def adicionar(self, pedido: Pedido) -> Pedido:
self._sessao.add(pedido)
self._sessao.flush()
return pedido
def obter(self, pedido_id: int) -> Pedido | None:
return self._sessao.get(Pedido, pedido_id)
def por_chave(self, chave: str) -> Pedido | None:
consulta = select(Pedido).where(Pedido.chave_idempotencia == chave)
return self._sessao.scalars(consulta).first()
def listar(self, depois_de: int, limite: int) -> list[Pedido]:
consulta = select(Pedido).where(Pedido.id > depois_de).order_by(Pedido.id).limit(limite)
return list(self._sessao.scalars(consulta))
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from api_pedidos.esquemas import PedidoCriar
from api_pedidos.modelos import ItemPedido, Pedido
from api_pedidos.repositorio import RepositorioPedidos
class PedidoNaoEncontrado(Exception):
def __init__(self, pedido_id: int) -> None:
super().__init__(f"pedido {pedido_id} não encontrado")
self.pedido_id = pedido_id
class PedidoJaCancelado(Exception):
def __init__(self, pedido_id: int) -> None:
super().__init__(f"pedido {pedido_id} já está cancelado")
self.pedido_id = pedido_id
class ServicoPedidos:
def __init__(self, sessao: Session) -> None:
self._sessao = sessao
self._repositorio = RepositorioPedidos(sessao)
def criar(self, dados: PedidoCriar, chave: str | None = None) -> tuple[Pedido, bool]:
"""Devolve o pedido e um indicador de criação (False quando a chave já existia)."""
if chave and (existente := self._repositorio.por_chave(chave)):
return existente, False
pedido = Pedido(
cliente=dados.cliente,
chave_idempotencia=chave,
itens=[ItemPedido(**item.model_dump()) for item in dados.itens],
)
try:
self._repositorio.adicionar(pedido)
self._sessao.commit()
except IntegrityError:
# Duas requisições com a mesma chave podem passar pela verificação acima ao mesmo
# tempo. Quem garante a unicidade é o banco, e aqui tratamos o erro dele.
self._sessao.rollback()
if chave and (existente := self._repositorio.por_chave(chave)):
return existente, False
raise
return pedido, True
def obter(self, pedido_id: int) -> Pedido:
pedido = self._repositorio.obter(pedido_id)
if pedido is None:
raise PedidoNaoEncontrado(pedido_id)
return pedido
def cancelar(self, pedido_id: int) -> Pedido:
pedido = self.obter(pedido_id)
if pedido.status == "cancelado":
raise PedidoJaCancelado(pedido_id)
pedido.status = "cancelado"
self._sessao.commit()
return pedido
def listar(self, cursor: int, limite: int) -> tuple[list[Pedido], int | None]:
# Busca um item a mais para saber se existe próxima página.
encontrados = self._repositorio.listar(cursor, limite + 1)
pagina = encontrados[:limite]
proximo = pagina[-1].id if len(encontrados) > limite else None
return pagina, proximo
Observabilidade
O middleware atribui o X-Request-ID, mede a duração, conta a requisição por rota padrão (baixa cardinalidade) e escreve uma linha de log em JSON:
import json
import logging
import sys
from collections.abc import Awaitable, Callable
from contextvars import ContextVar
from time import perf_counter
from uuid import uuid4
from fastapi import FastAPI, Request, Response
from prometheus_client import CONTENT_TYPE_LATEST, Counter, Histogram, generate_latest
id_requisicao: ContextVar[str] = ContextVar("id_requisicao", default="-")
REQUISICOES = Counter("api_requisicoes_total", "Total de requisições", ["metodo", "rota", "status"])
LATENCIA = Histogram("api_latencia_segundos", "Latência das requisições", ["rota"])
log = logging.getLogger("api")
class FormatoJson(logging.Formatter):
CAMPOS_EXTRAS = ("metodo", "rota", "status", "duracao_ms")
def format(self, record: logging.LogRecord) -> str:
dados: dict[str, object] = {
"nivel": record.levelname,
"mensagem": record.getMessage(),
"id_requisicao": id_requisicao.get(),
}
for campo in self.CAMPOS_EXTRAS:
if hasattr(record, campo):
dados[campo] = getattr(record, campo)
return json.dumps(dados, ensure_ascii=False)
def configurar_logs(nivel: str = "INFO") -> None:
manipulador = logging.StreamHandler(sys.stdout)
manipulador.setFormatter(FormatoJson())
log.handlers = [manipulador]
log.setLevel(nivel)
log.propagate = False
def instalar_observabilidade(app: FastAPI) -> None:
@app.middleware("http")
async def medir(
request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
identificador = request.headers.get("X-Request-ID") or uuid4().hex
token = id_requisicao.set(identificador)
inicio = perf_counter()
status = 500
try:
resposta = await call_next(request)
status = resposta.status_code
finally:
duracao = perf_counter() - inicio
rota_encontrada = request.scope.get("route")
rota = getattr(rota_encontrada, "path", "desconhecida")
REQUISICOES.labels(request.method, rota, str(status)).inc()
LATENCIA.labels(rota).observe(duracao)
log.info(
"requisição concluída",
extra={
"metodo": request.method,
"rota": rota,
"status": status,
"duracao_ms": round(duracao * 1000, 2),
},
)
id_requisicao.reset(token)
resposta.headers["X-Request-ID"] = identificador
return resposta
@app.get("/metricas", include_in_schema=False)
def metricas() -> Response:
return Response(generate_latest(), media_type=CONTENT_TYPE_LATEST)
A aplicação
A rota só traduz HTTP. Os erros do domínio viram problem+json em um único lugar, e a função criar_app recebe o engine como parâmetro, o que permite aos testes injetarem um banco descartável:
from typing import Annotated
from fastapi import Depends, FastAPI, Header, Query, Request, Response
from fastapi.responses import JSONResponse
from sqlalchemy import Engine, text
from sqlalchemy.orm import Session
from api_pedidos.config import obter_configuracao
from api_pedidos.db import criar_engine, criar_fabrica_de_sessoes, obter_sessao
from api_pedidos.esquemas import Pagina, PedidoCriar, PedidoLer
from api_pedidos.observabilidade import configurar_logs, instalar_observabilidade
from api_pedidos.servico import PedidoJaCancelado, PedidoNaoEncontrado, ServicoPedidos
SessaoDep = Annotated[Session, Depends(obter_sessao)]
def obter_servico(sessao: SessaoDep) -> ServicoPedidos:
return ServicoPedidos(sessao)
ServicoDep = Annotated[ServicoPedidos, Depends(obter_servico)]
def problema(status: int, titulo: str, detalhe: str) -> JSONResponse:
"""Formato de erro único para toda a API (inspirado no RFC 9457)."""
return JSONResponse(
{"titulo": titulo, "status": status, "detalhe": detalhe},
status_code=status,
media_type="application/problem+json",
)
def criar_app(engine: Engine | None = None) -> FastAPI:
config = obter_configuracao()
configurar_logs(config.log_nivel)
engine = engine or criar_engine(config.database_url.get_secret_value())
app = FastAPI(title="API de pedidos", version="0.1.0")
app.state.engine = engine
app.state.fabrica = criar_fabrica_de_sessoes(engine)
instalar_observabilidade(app)
@app.exception_handler(PedidoNaoEncontrado)
def nao_encontrado(_: Request, erro: PedidoNaoEncontrado) -> JSONResponse:
return problema(404, "Pedido não encontrado", str(erro))
@app.exception_handler(PedidoJaCancelado)
def ja_cancelado(_: Request, erro: PedidoJaCancelado) -> JSONResponse:
return problema(409, "Conflito de estado", str(erro))
@app.post("/pedidos", response_model=PedidoLer, status_code=201)
def criar_pedido(
dados: PedidoCriar,
servico: ServicoDep,
resposta: Response,
idempotency_key: Annotated[str | None, Header(max_length=80)] = None,
) -> object:
pedido, criado = servico.criar(dados, idempotency_key)
if not criado:
resposta.status_code = 200
return pedido
@app.get("/pedidos/{pedido_id}", response_model=PedidoLer)
def obter_pedido(pedido_id: int, servico: ServicoDep) -> object:
return servico.obter(pedido_id)
@app.get("/pedidos", response_model=Pagina)
def listar_pedidos(
servico: ServicoDep,
limite: Annotated[int, Query(ge=1, le=100)] = 20,
cursor: Annotated[int, Query(ge=0)] = 0,
) -> object:
pagina, proximo = servico.listar(cursor, limite)
return {"itens": pagina, "proximo_cursor": proximo}
@app.post("/pedidos/{pedido_id}/cancelar", response_model=PedidoLer)
def cancelar_pedido(pedido_id: int, servico: ServicoDep) -> object:
return servico.cancelar(pedido_id)
@app.get("/saude")
def saude(sessao: SessaoDep) -> JSONResponse:
try:
sessao.execute(text("SELECT 1"))
except Exception:
return JSONResponse({"banco": "indisponível"}, status_code=503)
return JSONResponse({"banco": "ok"})
return app
Os testes
Os testes de API usam o banco real (SQLite em memória por padrão). Defina TEST_DATABASE_URL para rodar a mesma suíte contra o PostgreSQL, e o que passa nos dois é o que dá confiança:
import os
from collections.abc import Iterator
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import Engine, create_engine
from sqlalchemy.pool import StaticPool
from api_pedidos.app import criar_app
from api_pedidos.modelos import Base
@pytest.fixture
def engine() -> Iterator[Engine]:
"""SQLite em memória por padrão. Defina TEST_DATABASE_URL para testar contra PostgreSQL."""
url = os.environ.get("TEST_DATABASE_URL")
if url:
eng = create_engine(url)
else:
eng = create_engine(
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
Base.metadata.drop_all(eng)
Base.metadata.create_all(eng)
yield eng
Base.metadata.drop_all(eng)
eng.dispose()
@pytest.fixture
def cliente(engine: Engine) -> TestClient:
return TestClient(criar_app(engine))
from fastapi.testclient import TestClient
CORPO = {
"cliente": "Ana",
"itens": [
{"produto": "caneta", "quantidade": 2, "preco_centavos": 350},
{"produto": "caderno", "quantidade": 1, "preco_centavos": 1890},
],
}
def test_criar_pedido(cliente: TestClient) -> None:
resposta = cliente.post("/pedidos", json=CORPO)
assert resposta.status_code == 201
dados = resposta.json()
assert dados["total_centavos"] == 2590
assert dados["status"] == "aberto"
assert len(dados["itens"]) == 2
def test_idempotencia_devolve_o_mesmo_pedido(cliente: TestClient) -> None:
cabecalho = {"Idempotency-Key": "abc-123"}
primeira = cliente.post("/pedidos", json=CORPO, headers=cabecalho)
segunda = cliente.post("/pedidos", json=CORPO, headers=cabecalho)
assert primeira.status_code == 201
assert segunda.status_code == 200
assert primeira.json()["id"] == segunda.json()["id"]
assert len(cliente.get("/pedidos").json()["itens"]) == 1
def test_validacao_recusa_corpo_invalido(cliente: TestClient) -> None:
resposta = cliente.post("/pedidos", json={"cliente": "", "itens": []})
assert resposta.status_code == 422
def test_pedido_inexistente_devolve_problema(cliente: TestClient) -> None:
resposta = cliente.get("/pedidos/999")
assert resposta.status_code == 404
assert resposta.headers["content-type"].startswith("application/problem+json")
assert resposta.json()["titulo"] == "Pedido não encontrado"
def test_cancelar_duas_vezes_devolve_conflito(cliente: TestClient) -> None:
pedido_id = cliente.post("/pedidos", json=CORPO).json()["id"]
assert cliente.post(f"/pedidos/{pedido_id}/cancelar").json()["status"] == "cancelado"
assert cliente.post(f"/pedidos/{pedido_id}/cancelar").status_code == 409
def test_paginacao_por_cursor(cliente: TestClient) -> None:
for _ in range(5):
cliente.post("/pedidos", json=CORPO)
primeira = cliente.get("/pedidos", params={"limite": 2}).json()
assert len(primeira["itens"]) == 2
assert primeira["proximo_cursor"] == 2
ultima = cliente.get("/pedidos", params={"limite": 2, "cursor": 4}).json()
assert [p["id"] for p in ultima["itens"]] == [5]
assert ultima["proximo_cursor"] is None
def test_saude_e_metricas(cliente: TestClient) -> None:
assert cliente.get("/saude").json() == {"banco": "ok"}
cliente.get("/pedidos")
assert "api_requisicoes_total" in cliente.get("/metricas").text
def test_id_de_requisicao_e_devolvido(cliente: TestClient) -> None:
resposta = cliente.get("/saude", headers={"X-Request-ID": "req-42"})
assert resposta.headers["x-request-id"] == "req-42"
O teste de serviço reproduz a corrida de duas requisições com a mesma chave: ele força a verificação a "não ver" o pedido existente, e prova que o IntegrityError do banco leva ao pedido original, em vez de um erro 500:
import pytest
from sqlalchemy import Engine
from sqlalchemy.orm import Session
from api_pedidos.esquemas import ItemCriar, PedidoCriar
from api_pedidos.modelos import Pedido
from api_pedidos.repositorio import RepositorioPedidos
from api_pedidos.servico import PedidoNaoEncontrado, ServicoPedidos
DADOS = PedidoCriar(
cliente="Ana", itens=[ItemCriar(produto="caneta", quantidade=1, preco_centavos=100)]
)
def test_obter_pedido_inexistente(engine: Engine) -> None:
with Session(engine) as sessao, pytest.raises(PedidoNaoEncontrado):
ServicoPedidos(sessao).obter(1)
def test_corrida_de_chaves_devolve_o_pedido_existente(
engine: Engine, monkeypatch: pytest.MonkeyPatch
) -> None:
with Session(engine) as sessao:
original, criado = ServicoPedidos(sessao).criar(DADOS, "chave-1")
assert criado
original_id = original.id
chamadas = {"total": 0}
por_chave_real = RepositorioPedidos.por_chave
def por_chave_atrasada(self: RepositorioPedidos, chave: str) -> Pedido | None:
chamadas["total"] += 1
if chamadas["total"] == 1:
return None # simula a outra requisição ainda não ter gravado
return por_chave_real(self, chave)
monkeypatch.setattr(RepositorioPedidos, "por_chave", por_chave_atrasada)
with Session(engine) as sessao:
pedido, criado = ServicoPedidos(sessao).criar(DADOS, "chave-1")
assert criado is False
assert pedido.id == original_id
O teste de migração garante que os modelos e as migrações não divergiram: ele aplica as migrações em um banco vazio e roda o equivalente a alembic check. Se alguém mudar um modelo sem criar a migração, a suíte falha:
from pathlib import Path
import pytest
from alembic import command
from alembic.config import Config
from api_pedidos.config import obter_configuracao
RAIZ = Path(__file__).resolve().parent.parent
def test_migracoes_estao_sincronizadas_com_os_modelos(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("API_DATABASE_URL", f"sqlite:///{tmp_path / 'migracao.db'}")
obter_configuracao.cache_clear()
cfg = Config(str(RAIZ / "alembic.ini"))
try:
command.upgrade(cfg, "head")
command.check(cfg) # falha se os modelos mudaram sem uma migração correspondente
finally:
obter_configuracao.cache_clear()
Rodar tudo
cd exemplos/api_pedidos
uv sync
uv run pytest
Saída
........... [100%]
11 passed
Contra o PostgreSQL, com a mesma suíte (a variável aponta para um banco de teste descartável):
TEST_DATABASE_URL="postgresql+psycopg://usuario:senha@localhost:5432/pedidos_teste" uv run pytest
E as verificações estáticas:
uv run mypy
uv run ruff check .
uv run ruff format --check .
Saída
Success: no issues found in 9 source files
All checks passed!
Usar a API
Aplique as migrações e suba o servidor. Com o Idempotency-Key, a segunda chamada devolve o pedido original, com status 200 (a primeira foi 201):
uv run alembic upgrade head
uv run uvicorn api_pedidos.app:criar_app --factory --port 8000
curl -s -X POST localhost:8000/pedidos \
-H "Idempotency-Key: k1" -H "Content-Type: application/json" \
-d '{"cliente": "Ana", "itens": [{"produto": "caneta", "quantidade": 2, "preco_centavos": 350}]}'
Saída
{"id":1,"cliente":"Ana","status":"aberto","total_centavos":700,"criado_em":"2026-10-06T15:48:21.782501Z","itens":[{"produto":"caneta","quantidade":2,"preco_centavos":350}]}
Um pedido inexistente devolve o erro no formato único, e cancelar duas vezes devolve 409:
GET /pedidos/9999 -> 404 {"titulo":"Pedido não encontrado","status":404,"detalhe":"pedido 9999 não encontrado"}
POST /pedidos/1/cancelar -> 200 (status "cancelado")
POST /pedidos/1/cancelar -> 409
As decisões, e o porquê
| Decisão | Alternativa | Por que escolhi assim |
|---|---|---|
| Dinheiro em centavos inteiros | float, ou Decimal | Sem erro de arredondamento e sem conversão no banco |
| Paginação por cursor | Por deslocamento | Estável sob escrita concorrente, e usa o índice da chave |
Idempotência com chave e UNIQUE | Só verificar antes | A verificação sozinha tem corrida. Quem garante é o banco |
Erro em problem+json | Texto livre por rota | O cliente trata por tipo, e não por texto |
| SQLAlchemy síncrono | async | O FastAPI roda rotas def em threads, e o driver síncrono é mais simples e basta para esta carga |
| Serviço sem FastAPI | Regra dentro da rota | Testável sem HTTP e reutilizável |
| Migração como etapa separada | Migrar na partida da API | Duas réplicas não disputam a migração |
O que quebra primeiro com dez vezes mais carga
Eu aplico a revisão que faço em qualquer serviço antes de aprová-lo:
- Conexões com o banco. O pool padrão do SQLAlchemy tem 5 conexões e 10 de excedente. Com muitas threads e consultas lentas, as requisições esperam por uma conexão. É o primeiro gargalo, e se mede observando o tempo de espera.
- Tabela de chaves de idempotência. Ela só cresce. Em produção, as chaves precisam expirar (uma rotina que apaga as antigas).
- Sem autenticação nem limite de requisições. Esta API é aberta. O próximo passo real é OAuth2 com JWT e um limitador de taxa.
- Sem tempo limite nas consultas. Uma consulta lenta segura uma conexão. Configure
statement_timeoutno PostgreSQL. - Eventos fora da transação. Se o pedido precisar avisar outro sistema, gravar no banco e publicar uma mensagem não é atômico. O padrão para isso é a outbox: gravar o evento na mesma transação e publicá-lo depois.
O que este projeto não prova
Os testes mostram que o código está correto e coerente com o banco. Eles não mostram que a imagem Docker constrói no seu ambiente nem que o sistema aguenta a carga real. Para isso, o passo seguinte é um teste de carga (com
locustouk6) contra um ambiente parecido com o de produção.