/
Chapter 62, Backend
Complete API: from contract to deploy
This chapter brings together everything the previous ones taught in a working end-to-end order API. Each file below has been executed: the tests pass against SQLite and against PostgreSQL, and strict mypy and ruff approve the code.
The files for this chapter are in exemplos/api_pedidos/.
What the API does
It is an order API with five routes, designed to show production decisions, not just routes that work:
| Route | What it does | Production detail |
|---|---|---|
POST /pedidos | Creates an order | Idempotency-Key prevents duplicate orders |
GET /pedidos/{id} | Reads an order | Error 404 in the format problem+json |
GET /pedidos | Lists with pagination | Cursor, not offset |
POST /pedidos/{id}/cancelar | Cancels | 409 if already canceled |
GET /saude and GET /metricas | Health and metrics | For the orchestrator and Prometheus |
The layered structure
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
The dependency arrow always goes from top to bottom: the route calls the service, the service calls the repository. The service does not care about FastAPI, and therefore it can be tested without HTTP and reused in a CLI or in a worker.
Dependencies and tools
The pyproject.toml declares the production dependencies, the development ones (in a separate group, which the Dockerfile does not install) and the configuration of pytest, ruff, and 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"]
Configuration and database
The configuration is from chapter 58, and the guard prod with SQLite prevents the most common deployment error:
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()
The model declares the database constraints: unique idempotency key, index by client, and foreign key with ON DELETE CASCADE. The money stays in whole cents:
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
The contract
The Pydantic schemas are the contract: they validate the input (positive quantity, at least one item) and define exactly what the output contains. FastAPI derives the OpenAPI documentation from them:
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
Repository and service
The repository only knows how to query. The service decides. Note in criar: the key check and the handling of IntegrityError coexist, because the check avoids work most of the time and the restriction UNIQUE of the database ensures correctness when two requests compete for the same key:
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
Observability
The middleware assigns the X-Request-ID, measures the duration, counts the request per route standard (low cardinality) and writes a line of log in 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)
The application
The route only translates HTTP. Domain errors turn into problem+json in one place, and the function criar_app receives the engine as a parameter, which allows tests to inject a disposable database:
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
The tests
API tests use the real database (SQLite in memory by default). Set TEST_DATABASE_URL to run the same suite against PostgreSQL, and what passes in both is what gives confidence:
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"
The service test reproduces the race of two requests with the same key: it forces the check to "not see" the existing request, and proves that the IntegrityError of the database leads to the original request, instead of an error 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
The migration test ensures that the models and migrations have not diverged: it applies the migrations to an empty database and runs the equivalent of alembic check. If someone changes a model without creating the migration, the suite fails:
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()
Run everything
cd exemplos/api_pedidos
uv sync
uv run pytest
Output
........... [100%]
11 passed
Against PostgreSQL, with the same suite (the variable points to a disposable test database):
TEST_DATABASE_URL="postgresql+psycopg://usuario:senha@localhost:5432/pedidos_teste" uv run pytest
And the static checks:
uv run mypy
uv run ruff check .
uv run ruff format --check .
Output
Success: no issues found in 9 source files
All checks passed!
Use the API
Apply the migrations and start the server. With the Idempotency-Key, the second call returns the original request, with status 200 (the first was 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}]}'
Output
{"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}]}
A non-existent request returns the error in a unique format, and canceling twice returns 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
The decisions, and why
| Decision | Alternative | Why I chose this way |
|---|---|---|
| Money in whole cents | float, or Decimal | No rounding error and no conversion in the database |
| Cursor pagination | By shifting | Stable under concurrent writes, and uses the key index |
Idempotency with key and UNIQUE | Just check beforehand | The check alone has a race condition. The database guarantees it |
Error in problem+json | Free text by route | The client handles by type, not by text |
| Synchronous SQLAlchemy | async | FastAPI runs routes def in threads, and the synchronous driver is simpler and sufficient for this load |
| Service without FastAPI | Rule within the route | Testable without HTTP and reusable |
| Migration as a separate step | Migrate at the API startup | Two replicas do not compete for migration |
What breaks first under ten times the load
I apply the review I do on any service before approving it:
- Connections with the database. The default SQLAlchemy pool has 5 connections and 10 excess. With many threads and slow queries, requests wait for a connection. It is the first bottleneck, and it is measured by observing the wait time.
- Idempotency key table. It only grows. In production, keys need to expire (a routine that deletes the old ones).
- No authentication or request limit. This API is open. The next real step is OAuth2 with JWT and a rate limiter.
- No timeout on queries. A slow query holds a connection. Configure
statement_timeoutin PostgreSQL. - Events outside the transaction. If the request needs to notify another system, writing to the database and publishing a message is not atomic. The standard for this is the outbox: record the event in the same transaction and publish it later.
What this project does not prove
The tests show that the code is correct and consistent with the database. They does not show that the Docker image builds in your environment nor that the system can handle the real load. For this, the next step is a load test (with
locustork6) against an environment similar to production.