backend-repo-architecture

Files (6)

• SKILL.md
• deployment-patterns.md · references/
• entrypoint-and-config.md · references/
• repo-evidence.md · references/
• structure-patterns.md · references/
• troubleshooting.md · references/

SKILL.md (first lines)

# Backend Repository Architecture

A canonical backend archetype for Python/FastAPI services, derived from real-world production systems.

## When to use

- Scaffolding a new FastAPI backend service
- Reorganizing an existing backend to match team conventions
- Understanding the standard service layout and responsibilities
- Choosing between flat vs versioned domain layouts
- Setting up multi-mode deployment (server + workers + consumers)
- Implementing the standard config/connection/routing patterns

## Core conventions

All conventions below are derived from production Python/FastAPI services:
- **Reference service A** (cleanest reference, server-only, SQLAlchemy 2.0, versioned domains)
- **Reference service B** (largest ~48 domain modules, flat layout, MODE dispatcher)
- **Reference service C** (monorepo: apps/ + packages/, MODE dispatcher)
- **Reference service D** (multi-deployment: SERVER_TYPE router gating, MODE dispatcher)

**Entrypoint dispatcher** (used by services B, C, D):
- `entrypoint.py` at repo root dispatches on `MODE` environment variable
- Supported modes: `server`, `consumer`, `temporal_worker`, `worker`, `cron`, `webhook_server`
- `server` mode runs `uvicorn app.application:get_app --factory=True`
- Worker/consumer modes call `asyncio.run(service_main())` on the respective service module
- Each mode may initialize metrics/telemetry before dispatching
- **Note**: Service A is server-only (no entrypoint.py, Dockerfile runs gunicorn directly)

**Application factory** (app/application.py):
- `get_app() -> FastAPI` constructs the app instance
- Sets `debug=settings.DEBUG`, `lifespan=lifespan`, `default_response_class=ORJSONResponse`
- Includes `redirect_slashes=False` to prevent ambiguous routing
- Adds CORS middleware with `allow_origins`, `allow_credentials`, `allow_methods`, `allow_headers`
- Includes `api_router` via `app.include_router(api_router)`
- May mount additional ASGI apps (e.g., `/metrics` for Prometheus)
- Optionally sets `root_path` for multi-deployment routing

**Router aggregation** (app/router.py or src/main.py):
- `api_router = APIRouter()` aggregates all domain routers
- Health checks registered as `/_healthz` and `/_readyz`
- Domain routers included under version prefix (e.g., `/v1`)
- Routers may be conditionally registered based on `DEPLOYMENT_NAME` or `SERVER_TYPE` (multi-deployment variant)
- Uses `include_router(domain_router, prefix="/v1", tags=["Domain"])`

**Custom routing** (app/routing.py):
- `CustomRequestRoute(APIRoute)` provides centralized request/response logging, timing, exception handling
- Wraps all validation/HTTP/unhandled exceptions into a `ResponseData.error()` envelope
- Logs every request with method, path, status_code, duration
- Returns `ORJSONResponse` with consistent error format

**Connection management** (app/connection.py):
- `ConnectionManager` singleton (via `__new__` or Singleton metaclass) manages `AsyncEngine`, scoped session factory, and Redis client
- `async_scoped_session(sessionmaker(engine, expire_on_commit=False, class_=AsyncSession), scopefunc=current_task)`
- `ConnectionHandler` per-request wrapper provides lazy `.session` and `.redis` properties
- Supports RLS via `set_tenant_context(tenant_id)` and schema switching via `set_schema_context(schema_name)`
- `get_connection_handler()` dependency yields a handler and calls `handler.close()` on teardown

**Lifetime hooks** (app/lifetime.py):
- `@asynccontextmanager async def lifespan(app: FastAPI)` manages startup/shutdown
- Startup: `setup_logging()`, `ConnectionManager()`, Kafka producer, Temporal client, Redis
- Shutdown: close ConnectionManager, dispose engine, close Redis, shutdown telemetry

**Base DAO** (app/dao.py):
- `BaseDao` provides `create`, `get_by_pk`, `update_by_pk`, `delete_by_pk`, `get_by_filter`, `get_paginated_response`, `bulk_insert`, `bulk_update`
- Soft-delete aware: checks `is_deleted` and `deleted_at` attributes
- All queries go through `_execute_query(query)` which rolls back on error
- Pagination supports `page_size`, `page_number`, `sort_by`, `order_by` with column validation

**Database base** (app/database.py):
- SQLAlchemy 1.4 repos: `Base = declarative_base()`
- SQLAlchemy 2.0 repos: `class Base(DeclarativeBase): pass`
- All ORM models inherit from `Base`

**Config pattern** (config/settings.py or src/config/docker_config.py):
- `Settings(BaseSettings)` with pydantic-settings for typed environment variables
- Pattern A: single `Settings` class, module-level `settings = Settings()`
- Pattern B: `loaded_config = Settings()` singleton, plus `config_parser.py` for configargparse-based YAML/env merging
- All services consume `DATABASE_URL`, `REDIS_URL`, `DEBUG`, `CORS_ORIGINS`, `KAFKA_BOOTSTRAP_SERVERS`, `TEMPORAL_HOST`, etc.

**Domain layout: flat variant**:
- Top-level `src/<domain>/` for many domain modules
- Each module has `router.py`, `service.py`, `dao.py`, `models.py`, `schemas.py`, `helper.py`, `constant.py`
- Single-file per concern (no subdirectories within a domain)

**Domain layout: versioned variant**:
- `<pkg>/<domain>/v1/{router,service,dao,models,schemas,helper,workflows}.py`
- Allows side-by-side v1/v2 implementations

**Monorepo variant**:
- `apps/` (api, orchestrator, workers, forwarders)
- `packages/` (common, contracts, db, sdk)
- `services/` (infra packages like Kafka, Temporal)
- Each app has its own `application.py`, `router.py`, `lifetime.py`

**Multi-deployment variant**:
- Single codebase, `SERVER_TYPE` env var gates router inclusion (public/internal/platform/webhook)
- `custom_openapi()` rewrites paths with `root_path` like `/service/{SERVER_TYPE}/product/feature`
- Conditional `include_router()` based on `loaded_config.SERVER_TYPE`

## Skeleton / example

```
backend/
├── entrypoint.py              # MODE dispatcher (server/consumer/temporal_worker/cron) — OPTIONAL, omit for server-only
├── alembic/                   # DB migrations
├── alembic.ini
├── pyproject.toml
├── Dockerfile
├── app/                       # (or src/)
│   ├── application.py         # get_app() factory
│   ├── router.py              # api_router aggregation
│   ├── routing.py             # CustomRequestRoute (optional but recommended)
│   ├── connection.py          # ConnectionManager, ConnectionHandler, get_connection_handler
│   ├── dao.py                 # BaseDao
│   ├── database.py            # Base (declarative base)
│   ├── lifetime.py            # lifespan context manager
│   ├── main.py                # uvicorn runner (server mode)
│   ├── telemetry.py           # OpenTelemetry instrumentation (optional)

From ajyadav013/claude-kit, MIT.

More from ajyadav013/claude-kit