Inside the AtlasboxPy API Stack

One call path was only the first piece

An earlier post walked through the actual bug that started all of this: a PATCH /users/{id} endpoint that wiped a user's last name because Optional[str] = None can't tell "the client cleared this" from "the client never sent this." The fix for that one bug is a keyword. The fix for the pattern underneath it — business-rule validation, error formatting, and response shaping quietly becoming the HTTP layer's job — is atlasboxpy_controller's BaseController: one call path, one error taxonomy, for a REST route, a worker, or an agent alike.

But a controller that enforces one call path still has to call something. Once validation and error handling stop being the router's problem, the very next question is where the data actually lives, how it's cached, and how a controller orchestrating more than one service stays fast instead of just correct. That's the rest of the stack this post covers — walked through with the same worked example the packages themselves ship: a small Kanban board.

The three packages that make a stack

AtlasboxPy is a workspace of small, independent, flat-named Python packages — each its own installable package, no shared namespace between them. Three of them are the actual core of a typical application:

Building something that talks to a database and might need caching?
                              │
        ┌─────────────────────┼─────────────────────┐
        │                     │                      │
   Need one consistent   Need a pluggable,      Need to configure
   response shape        read-through cache     which physical
   across REST/worker/   in front of your       database (or shard)
   agent callers, with   data access, without   a table lives in,
   error handling that's hand-rolling            decoupled from the
   structural, not       get/set/invalidate      query logic that
   opt-in per route?     plumbing?               uses it?
        │                     │                      │
        ▼                     ▼                      ▼
   atlasboxpy_controller  atlasboxpy_repository  atlasboxpy_db
   (BaseController)       (BaseRepository)       (DBQuantum/ShardRouter)
        │                     │                      │
        └─────────────────────┼──────────────────────┘
                               ▼
             Most real apps use all three together.

atlasboxpy_repository gives you a BaseRepository with a pluggable, read-through cache — swap between an in-memory dict and Redis via two config constants. atlasboxpy_db gives you DBQuantum and ShardRouter for SQLAlchemy: a single physical database is just a ShardRouter with one shard, not a separate type from a sharded one, so going to N databases later is a config change, not a migration. It also has a VariantRouter, for picking between a small set of semantically different databases — a shadow database seeded with test data for post-deploy validation, say — by an exact label, deliberately not the same mechanism as sharding.

One base class, one job

Every package above hands you exactly one base class or one mechanism, and each one does exactly one thing. None of them inherit from each other — they compose, through configuration, not a shared class hierarchy:

BaseController              (atlasboxpy_controller)
  job: one call path, one error taxonomy
  ExceptionFormatter wraps every public async method automatically
      │
      ▼ subclassed by
  KanbanController, UserController, SecurityFindingController, ...


BaseRepository              (atlasboxpy_repository)
  job: owns self.cache — nobody above it knows a cache exists
      │
      ▼ subclassed by
  BoardRepository, ColumnRepository, CardRepository, ...
      │
      ▼ self.cache is one of, chosen by cache_driver / cache_env
  BareMetalCacheBackend (in-memory)     RedisCacheBackend (shared, TTL)


DBQuantumRegistry            (atlasboxpy_db)
  job: resolve "which session for which database"
      │
      ▼ hands out sessions through
  ShardRouter   — one or many DBQuantum shards, same call either way
  VariantRouter — prod vs. shadow, exact match, safe default on anything else


HeaderContextMiddleware + RequestContext     (atlasboxpy_api)
  job: turn one HTTP header into request-scoped state, leak-proof
      │
      ▼ read by
  Tracer                       (atlasboxpy_telemetry)
  job: one trace id per request, built on the same context above

Read top to bottom, that's the whole stack's job list in one place: a controller enforces a call path, a repository owns its cache, a registry resolves a session, a header becomes request-scoped state, and a tracer rides on top of that same state. Nothing here inherits behavior it doesn't need — a repository never learns HTTP exists, and a controller never learns SQL exists.

The controller, orchestrating one service

KanbanController subclasses BaseController and orchestrates KanbanService — nothing more. It never references a persistence-layer type; that's each entity repository's concern, several layers down. Every public method is wrapped in a try/except automatically, at class-definition time, and each takes exactly one argument — props, a plain dict — which it validates against the matching model in models.py. That model is the method's contract:

class KanbanController(BaseController):
    def __init__(self) -> None:
        super().__init__()
        self.service = KanbanService()

    async def create_card(self, props: dict[str, Any]) -> SuccessResponse[Any] | ErrorResponse:
        payload = validate_props(CreateCardRequest, props)  # {board_id, column_id, title, description}
        response = self._response_for(
            await self.service.create_card(
                payload.board_id, payload.column_id, payload.title, payload.description
            )
        )
        return self._with_card_title_hint(response)

    async def move_card(self, props: dict[str, Any]) -> SuccessResponse[Any] | ErrorResponse:
        payload = validate_props(MoveCardRequest, props)  # {card_id, column_id}
        response = self._response_for(await self.service.move_card(payload.card_id, payload.column_id))
        if isinstance(response, SuccessResponse):
            # A move is a domain event, not just a data read — mark it that way so a
            # caller can tell the two apart from status/response_code alone.
            return SuccessResponse(status=ResponseStatus.EVENT_FIRED, response_code=202, data=response.data)
        return response

create_card never raises and never talks HTTP directly — it builds one typed, transport-agnostic response the same way regardless of who's calling. And because validation lives in the method next to its model instead of in a separate route file, reading create_card alone tells you everything a call needs: board_id, column_id, title, description — nothing left implicit somewhere else.

Repositories and Entity-Type Storage

BoardRepository, ColumnRepository, and CardRepository each subclass BaseRepository and are the only places in this chain that touch a cache — KanbanService and KanbanController never see it, and none of the three repositories knows the other two exist. Each independently-accessed entity (Board, Column, Card — plain dataclasses) gets its own storage class naming only the operations that entity needs, with its own connection config:

class SQLAlchemyCardStorage:
    def __init__(self, sessions: SessionOpener) -> None:
        self._sessions = sessions

    async def create(self, board_id: str, column_id: str, title: str, description: str) -> Card:
        card_id = str(uuid.uuid4())
        async with session_scope(self._sessions) as session:
            session.add(CardRow(id=card_id, board_id=board_id, column_id=column_id,
                                 title=title, description=description))
        return Card(id=card_id, board_id=board_id, column_id=column_id,
                     title=title, description=description)

That buys five things:

• Cache invalidation stays with the entity that owns the data. Each repository only ever invalidates its own cache key — adding a card no longer busts the columns cache the way one shared "the whole board" cache entry used to.
• An entity's database is a config change, not a refactor. Card — by far the highest write volume — can move onto its own database instance by editing its connection config, nothing else. DBQuantumRegistry caches engines and session factories by quantum name, so entities that still share one also still share a connection pool.
• Callers see application types, never a SQLAlchemy row. Storage classes return plain dataclasses, not ORM rows — a session-bound row is cache-unsafe the moment its session closes; a dataclass isn't.
• The trade-off is explicit, not hidden. Multi-entity operations (a board plus its default columns) no longer get a free atomic transaction once entities can live on different quanta — that cost is documented, not discovered in production.
• A non-SQL backend that's been rejected says so, out loud. Not every entity fits SQLAlchemy: an append-only activity log gets its own Cassandra-backed quantum instead of forcing a relational index onto a partition-and-scan access pattern. And when a backend was actually considered and turned down — MongoDB, in this example — its constructor doesn't just not exist; it raises with the real reasoning attached, so a developer reaching for it later hits the decision, not silence.

The concurrent read that makes it pay off

KanbanService.get_board — the one expensive assembled read, a board with its columns and every card nested inside — is where those three independent caches actually pay off: each repository call is its own read-through cache, and all three run concurrently since none depends on the others.

class KanbanService:
    def __init__(self) -> None:
        self._boards = BoardRepository()
        self._columns = ColumnRepository()
        self._cards = CardRepository()

    async def get_board(self, board_id: str) -> ServiceResult:
        board, columns, cards = await asyncio.gather(
            self._boards.get_by_id(board_id),        # its own cache: kanban:board:{id}
            self._columns.list_for_board(board_id),   # its own cache: kanban:columns:{board_id}
            self._cards.list_for_board(board_id),     # its own cache: kanban:cards:{board_id}
        )
        if board is None:
            return ServiceResult.error(f"Board {board_id} not found", code="not_found")
        # ... assemble board + columns + cards into one dict ...
        return ServiceResult.ok(data)

Swapping a repository's cache from an in-memory dict to Redis is a two-constant change, made independently per entity — KanbanService and KanbanController never know a cache exists, let alone which technology backs which entity. Keeping the cache in-process is right for a single instance or an agent running locally; pointing it at a shared Redis URL makes it a cache every node or replica actually shares — and since that's an environment variable, relocating it to a different cloud account, region, or network-isolated instance is a config change, not a code change.

Configuration decides, code doesn't

Every decision above — which cache backs a repository, how many shards a database has, which environment a request talks to, whether a call chain gets traced — is a declared constant, an environment variable, or a header value. None of them is a branch an engineer (or an agent) has to write and get right every time.

• Two enum constants pick the cache backend. A repository declares cache_driver: CacheDriver (BARE_METAL or REDIS) and cache_env: CacheEnv (LOCAL or REMOTE) at the top of its own file. Both backends implement the same get/set/invalidate contract, so the repository's own query methods never branch on which one is active — flipping the constant is the entire migration. The one sharp edge this doesn't hide: a BARE_METAL cache is process-local, so a second OS process writing the same row can leave a BARE_METAL-cached read stale elsewhere. The fix is per-repository — drop caching on that specific read, or move it to REDIS — never a package-wide change.
• A shard list is the only difference between one database and twelve. ShardRouter(name="kanban", shards=[DBQuantum(...)]) is the same call whether that list has one entry or twelve. Going from one physical database to a sharded one is adding entries to a list, not rewriting an "unsharded" code path into a "sharded" one.
• A header, resolved through a safe-default router, tests a debug route in production with no redeploy. HeaderContextMiddleware reads one HTTP header (say X-DB-Environment) into request-scoped state — a ContextVar, not a global, reset in a finally block even if the request raises. It has zero opinion about what the header means; pairing it with VariantRouter (exact match, safe default on anything unrecognized) is what turns "some string a client sent" into "prod" or "shadow" and nothing else. A request carrying X-DB-Environment: shadow gets routed to a shadow database seeded with test data, for that one request only — every other concurrent request, including ones hitting the same route with no header at all, still goes to prod.
• Tracing turns on two ways, neither of them a code change. A process-wide default lives in one environment variable (ATLASBOXPY_TELEMETRY_ENABLED); a per-request override rides the same header-plus-VariantRouter mechanism above, so a single production request can get a fully traced call chain — trace id, parent/child spans, logged as structured lines, no external tracing backend required — without flipping the process-wide switch for every other concurrent request.

That pattern is precisely what makes this safe for an agent to extend. Adding a new repository, a new shard, or a new traced request doesn't mean writing new control flow and hoping it's reviewed carefully enough — it means setting the same two constants, appending to the same list, or reading the same header the existing code already reads. The established pattern is the guardrail; following it is what an agent-driven change to production actually looks like here.

One standardized response, read two ways

Every BaseController method returns the same envelope, and that envelope carries its own transport-agnostic verdict — not just a success/error binary, but a status label (success, event-fired, error, timeout, not-found, exception, api-error, out-of-memory, stack-overflow) plus a numeric response code, decided once at the controller from the DomainError raised or built — never duplicated per transport:

class SuccessResponse(BaseModel, Generic[T]):
    status: ResponseStatus = ResponseStatus.SUCCESS
    response_code: int = 200
    data: T

class ErrorResponse(BaseModel):
    status: ResponseStatus = ResponseStatus.ERROR
    response_code: int = 500
    error: ErrorDetail

The point isn't to hide the HTTP status code — the JSON-response helper still sets a real one on the wire — it's that an in-process caller never has to make an HTTP call to find out what that status code would have been. An agent holding a controller instance reads result.status/result.response_code straight off the object it already has, no handshake, no client, no schema to parse.

Two callers, same controller method, same envelope:

# REST route — extracts props from the request, calls the controller, converts
# the result to a JSONResponse using result.response_code as the actual HTTP status.
async def move_card(request: Request) -> JSONResponse:
    props = await extract_api_request(request)
    result = await controller.move_card(props)
    return to_json_response(result)

# AI agent — no HTTP at all. Holds the controller instance, calls the method as
# a plain coroutine, and branches on the same finite status vocabulary either way.
result = await controller.move_card({"card_id": card_id, "column_id": "col-doing"})

match result.status:
    case ResponseStatus.SUCCESS | ResponseStatus.EVENT_FIRED:
        ...  # use result.data
    case ResponseStatus.TIMEOUT:
        ...  # transient — is_retryable(result.error.code) confirms a retry can help
    case ResponseStatus.NOT_FOUND | ResponseStatus.ERROR:
        ...  # surface result.error.message, maybe ask a clarifying question
    case ResponseStatus.EXCEPTION | ResponseStatus.OUT_OF_MEMORY | ResponseStatus.STACK_OVERFLOW:
        ...  # a real bug, not a business outcome — don't retry blindly, flag it

The agent doesn't need a REST client, an OpenAPI schema, or HTTP-status guessing to know what happened, and doesn't need a running server to develop or test against this — it calls the same method a route calls, with the same one-dict argument, in-process, and gets back a typed, machine-checkable verdict either way. A move is worth calling out specifically here: it gets its own event-fired status and a 202 response code instead of a plain success, because a card changing columns is a domain event, not just a data read — and an outage on a read can come back as a clearly-marked degraded 200-adjacent response instead of an error, when returning something is better than returning nothing.

The narrower add-ons

atlasboxpy_api, atlasboxpy_service, and atlasboxpy_telemetry aren't part of the core three — each is a narrower add-on for a specific need that a real app reaches for once the core three are already in place, not before. The header-context and tracing mechanics for the first and third are covered above; the one still worth naming on its own is atlasboxpy_service:

• atlasboxpy_service — a BaseService base class for the layer that orchestrates repositories and third-party or internal-service calls on a controller's behalf: the same auto-wrapping mechanism as BaseController, logging every call's entry and outcome, plus a reusable named-concurrent-call helper for orchestrating more than one service at once.

How this scales

The packages ship with runnable example apps of increasing scope, each one literally branched from the last:

• fastapi_kanban — the focused demo of the core three, walked through above: Starlette plus SQLite, real cache invalidation, consistent error handling.
• fastapi_agile_project_planner — starts as the same app, with atlasboxpy_service, atlasboxpy_telemetry, and atlasboxpy_api wired in, and a controller orchestrating three services (a user-session service, a kanban service, and a task-agent service), each owning one bounded concern and never calling each other directly.

The pattern is the guardrail

One enforced call path through BaseController is the piece that gets the most attention, because it's the piece that shows up the moment a worker or an agent needs the same logic a REST route already has. But it's load-bearing precisely because everything underneath it — entity-scoped repositories, a read-through cache that never leaks into the service or controller layer, a database router that turns "add a shard" into a config change — stays just as disciplined. Every one of those base classes does exactly one job, and every decision built on top of them is a constant, an environment variable, or a header, not a branch. That's what actually makes the pattern followable: an agent extending this stack toward production isn't asked to reason about a tangle of conditionals, it's asked to declare the same two constants, append to the same list, or read the same header the rest of the codebase already does.

The Validation Layer That's Trapped Inside Your HTTP Framework covers the controller side of this in ADR form. MCP is for Strangers covers where the same controller sits relative to an actual protocol boundary. This post is the piece in between: what the controller is actually calling, and why that part staying disciplined is what makes the whole stack fast, cache-correct, and safe to hand to an agent.

Placeholder — content pending

Real benchmark numbers and screenshots of the DevSecOps Nanobar Admin application are planned follow-ups, not written yet.

Get the code

Every example walked through above — the Kanban controller, the entity-scoped repositories, the concurrent read — is runnable, not illustrative. See the full workspace on GitHub, including the examples/ directory these sections were drawn from.

Get packaged downloads