Skip to main content

Extending

Things the template does not include, on purpose, with the shortest path to add each. The recurring theme: add it as an adapter or a port, never inside domain/ or application/.

Authentication​

Shipped: API keys (shared/http/auth.py), Actor (shared/application/actor.py), and the "only the organizer can start" rule in StartTournament. See ADR-009.

API_KEYS='{"s3cret": "alice", "adm1n": "root:admin"}' make run
curl -X POST localhost:8000/api/v1/tournaments -H 'X-API-Key: s3cret' ...

To switch to JWT: replace get_actor in shared/http/auth.py with one that verifies the bearer token and builds an Actor from its claims. Nothing else changes.

Background jobs / outbox​

The in-process bus awaits handlers inside the request. When a handler becomes slow or unreliable (e-mail, third-party API):

  1. Add an outbox table and an OutboxEventPublisher adapter that inserts events in the same session as the aggregate (the port stays EventPublisher).
  2. Add a worker entrypoint (cleanarch/worker.py) that polls the outbox and pushes to a broker or calls the handlers.
  3. Change one line in bootstrap. Use cases do not change.

See ADR-005 for when this becomes worth it.

Unit of Work​

When one request must modify two aggregates atomically, or a use case must decide when to commit, introduce a UnitOfWork Protocol in shared/application/ports.py with __aenter__/__aexit__/commit/rollback and repository attributes, implement it in shared/infrastructure/, and pass it to those use cases instead of a repository. ADR-004 explains why it is not there by default.

Read models / CQRS​

If listing endpoints outgrow repository.list() (joins, aggregations, search), add a TournamentReadModel Protocol with query-shaped methods returning plain DTOs, implemented directly with SQL in infrastructure/. The write side keeps the repository. You have CQRS without a bus. ADR-003.

Caching​

A cache is an adapter that wraps another adapter:

class CachedTournamentRepository:
    def __init__(self, inner: TournamentRepository, cache: Redis) -> None: ...

Same Protocol, decorated in bootstrap.

Observability​

Shipped:

  • Request ids - shared/http/request_id.py reads or generates X-Request-ID, echoes it in every response and exposes it through a ContextVar.
  • Structured logs - bootstrap/logging.py: LOG_FORMAT=json emits one JSON object per line with time, level, logger, message, request_id; text for humans. Uvicorn's own loggers are routed through the same handler.
  • Probes - /health (liveness) and /ready (runs SELECT 1, 503 when the database is unreachable).

Next steps when you need them: OpenTelemetry's FastAPI and SQLAlchemy instrumentations are applied in create_app and make_engine; Prometheus metrics via a middleware in shared/http/. Nothing inside the rings changes.

Another driving adapter (consumer, gRPC, scheduler)​

The CLI in tournaments/cli/ + bootstrap/cli.py is the worked example: build the ports, open a transaction, call the use case, present the result. A message consumer follows the same shape with a loop around it. The architecture tests fail on a folder they do not know: add consumers to RING in scripts/archcheck.py with position 2 (same ring as http and cli) and the rule applies to it too.

Multiple features talking to each other​

Preferred: through events. Feature B subscribes to Feature A's events in bootstrap. Acceptable: Feature B's use case depends on a port that bootstrap implements by calling Feature A's use case. Forbidden: importing across features (the architecture tests fail).