design-patterns-and-conventions

Files (10)

• SKILL.md
• anti-patterns.md · references/
• htn-design-a-vending-machine.md · references/
• htn-lld-concepts.md · references/
• htn-validation-transformation-controllers-middlewares.md · references/
• naming-and-layout.md · references/
• patterns-catalog.md · references/
• repo-evidence.md · references/
• testing-patterns.md · references/
• troubleshooting.md · references/

SKILL.md (first lines)

Enforce consistent design patterns, naming conventions, and architectural decisions across FastAPI backends, derived from real-world production services.

## When to use

- Architecting new FastAPI services or major features
- Reviewing code structure and organization
- Establishing conventions for a new backend module
- Refactoring existing code for consistency
- Investigating or preventing technical debt
- Questions about layered architecture, dependency injection, or configuration management
- Setting up singleton instances, factory patterns, or custom API routes
- Organizing multi-mode entrypoints (server/consumer/worker in one image)
- Evaluating or auditing code for anti-patterns

## Core conventions

### Layered architecture
- **Router/View → Service → Helper → DAO/Repository → Model** is the canonical flow. Router handles HTTP concerns, service contains business logic, DAO/repository owns data access.
- Service layer MAY be thin or absent for pure CRUD operations where the router delegates directly to a DAO.
- Never mix business logic into routers or DAOs; keep separation strict.

### Dependency injection
- Use FastAPI `Depends()` for injecting context (database sessions, configuration, event emitters).
- Example pattern: `async def get_connection_handler_for_app() -> AsyncGenerator[ConnectionHandler, None]` yields a handler, then closes it in `finally`.
- NO dependency injection container; favor function-scoped generators and `Depends`.

### Configuration singleton
- Import a single `loaded_config` instance (a Pydantic `BaseSettings`) everywhere.
- Example: `from config.docker_config import loaded_config` then access `loaded_config.POSTGRES_READ_WRITE`.
- Settings are loaded once at startup; environment variables drive configuration.

### Singleton metaclass
- For truly global, single-instance objects (ConnectionManager, AsyncEventBridge), use a `Singleton` metaclass:
  ```python
  class Singleton(type):
      _instances = {}
      def __call__(cls, *args, **kwargs):
          if cls not in cls._instances:
              cls._instances[cls] = super().__call__(*args, **kwargs)
          return cls._instances[cls]
  ```
- Apply via `class ConnectionManager(metaclass=Singleton)`. Do NOT create duplicate instances elsewhere.

### Factory pattern
- Top-level factory functions: `get_app()` or `create_app()` constructs the FastAPI instance, applies middleware, wires routers, and returns it.
- Separate factory logic from the application object; enables clean testing and multi-mode entrypoints.

### Custom API route for cross-cutting concerns
- Centralize request/response logging, error handling, and envelope wrapping in a custom `APIRoute` class.
- Example: `CustomRequestRoute` inherits `fastapi.routing.APIRoute`, overrides `get_route_handler()`, wraps the original handler with try/except, logs request/response JSON, and returns a uniform `ResponseData` envelope.
- Apply globally: `api_router = APIRouter(route_class=CustomRequestRoute)`.

### Response envelope
- Return a uniform JSON shape from all endpoints: `{"success": bool, "data": Any, "message": str, "errors": list[str]}`.
- Use a Pydantic model `ResponseData` with class methods `.ok(data, message)` and `.error(errors, message)`.
- Custom route handlers catch exceptions and always return this envelope, even on validation errors or 500s.

### Event-driven architecture
- Kafka producer/consumer with **config-driven topic-to-handler maps**. The consumer reads a config mapping topic names to handler functions; no hardcoded dispatch logic.
- Favor async Kafka clients (aiokafka) for async FastAPI services; ensure producer is a singleton and consumer runs in its own process/MODE.

### Workflow orchestration
- Use Temporal for long-running, stateful workflows.
- A reference service includes a **JSON-DAG DSL interpreter** that turns declarative workflow JSON into Temporal execution graphs (config-as-data pattern).
- Workflow definitions are decoupled from business code.

### Multi-mode entrypoint
- One Docker image, many roles: server (FastAPI), consumer (Kafka), worker (Temporal), cron.
- `MODE` environment variable selects the main function in `entrypoint.py`:
  ```python
  if loaded_config.MODE == "server":
      server_main()
  elif loaded_config.MODE == "consumer":
      consumer_main()
  elif loaded_config.MODE == "temporal_worker":
      worker_main()
  ```
- Routers conditionally register based on MODE to reduce attack surface.

### Naming conventions
- **Files and functions:** `snake_case` (e.g., `connection_handler.py`, `get_session_factory()`).
- **Classes:** `PascalCase` (e.g., `ConnectionManager`, `ResponseData`).
- **Folders:** snake_case or hyphen-separated (`global_utils/`, `data_category_field/`).

### Folder layout
- **Per-domain modules:** Organize by domain/feature, not by layer. E.g., `src/supplier/`, `src/webhook/`, each with its own `router.py`, `service.py`, `dao.py`, `models.py`.
- **Top-level structure:**
  - `app/` or `src/` for application code
  - `config/` for settings, logging, uri parsing
  - `services/` for infrastructure integrations (Kafka, Temporal, GCS)
  - `global_utils/` or `core/` for shared utilities (metaclasses, exceptions, HTTP clients)
  - `alembic/` or `migrations/` for DB migrations
- **API versioning:** `v1/`, `v2/` as path prefixes; may group all v1 routers under `api/v1/router.py`.

## Skeleton / example

```python
# app/connection.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_scoped_session
from sqlalchemy.orm import sessionmaker
from asyncio import current_task
from global_utils.metaclasses import Singleton
from config.docker_config import loaded_config

class ConnectionManager(metaclass=Singleton):
    def __init__(self):
        self._db_engine, self._db_session_factory = self._setup_db()

    @staticmethod
    def _setup_db():
        db_url = str(loaded_config.POSTGRES_READ_WRITE)
        async_db_url = db_url.replace("postgresql://", "postgresql+asyncpg://")
        engine = create_async_engine(async_db_url, echo=loaded_config.DB_ECHO, pool_pre_ping=True)
        session_factory = async_scoped_session(
            sessionmaker(engine, expire_on_commit=False, class_=AsyncSession),
            scopefunc=current_task,
        )
        return engine, session_factory

From ajyadav013/claude-kit, MIT.

More from ajyadav013/claude-kit