Backend Module Development¶

Creating a New Module¶

This guide explains how to create a new feature module following Substrate's conventions.

Module Template¶

Create a new directory under app/modules/{module_name}/:

modules/
└── {module_name}/
    ├── __init__.py
    ├── router.py
    ├── service.py
    ├── repository.py
    ├── schemas.py
    ├── models.py          # Optional: SQLAlchemy models
    └── dependencies.py

File Templates¶

router.py¶

"""{Module} API routes."""
from fastapi import APIRouter, Depends

from app.core.security import UserInfo, get_current_user
from app.core.responses import list_response, envelope_response
from app.modules.{module}.dependencies import get_{module}_service
from app.modules.{module}.service import {Module}Service
from app.modules.{module}.schemas import (
    {Module}ListResponse,
    {Module}DetailResponse,
    Create{Module}Request,
)

router = APIRouter(prefix="/{modules}", tags=["{modules}"])


@router.get("", response_model={Module}ListResponse)
async def list_{modules}(
    service: {Module}Service = Depends(get_{module}_service),
    user: UserInfo = Depends(get_current_user),
):
    """List all {modules}."""
    items = await service.list_{modules}()
    return list_response(items)


@router.get("/{id}", response_model={Module}DetailResponse)
async def get_{module}(
    id: str,
    service: {Module}Service = Depends(get_{module}_service),
    user: UserInfo = Depends(get_current_user),
):
    """Get a specific {module} by ID."""
    item = await service.get_{module}(id)
    return {"data": item}


@router.post("", response_model={Module}DetailResponse, status_code=201)
async def create_{module}(
    request: Create{Module}Request,
    service: {Module}Service = Depends(get_{module}_service),
    user: UserInfo = Depends(get_current_user),
):
    """Create a new {module}."""
    item = await service.create_{module}(request, user)
    return {"data": item}

service.py¶

"""{Module} business logic."""
from app.modules.{module}.repository import {Module}Repository
from app.modules.{module}.schemas import Create{Module}Request, {Module}Out
from app.core.security import UserInfo


class {Module}Service:
    def __init__(self, repository: {Module}Repository):
        self._repo = repository

    async def list_{modules}(self) -> list[{Module}Out]:
        """List all {modules}."""
        return await self._repo.list_all()

    async def get_{module}(self, id: str) -> {Module}Out | None:
        """Get a {module} by ID."""
        return await self._repo.get_by_id(id)

    async def create_{module}(
        self,
        request: Create{Module}Request,
        user: UserInfo,
    ) -> {Module}Out:
        """Create a new {module}."""
        return await self._repo.create(request, user.sub)

repository.py¶

"""{Module} data access."""
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, insert

from app.modules.{module}.schemas import {Module}Out, Create{Module}Request
from app.modules.{module}.models import {Module}Model


class {Module}Repository:
    def __init__(self, session: AsyncSession):
        self._session = session

    async def list_all(self) -> list[{Module}Out]:
        """List all {modules}."""
        result = await self._session.execute(
            select({Module}Model).order_by({Module}Model.created_at.desc())
        )
        return [{Module}Out.model_validate(row) for row in result.scalars()]

    async def get_by_id(self, id: str) -> {Module}Out | None:
        """Get a {module} by ID."""
        result = await self._session.execute(
            select({Module}Model).where({Module}Model.id == id)
        )
        row = result.scalar_one_or_none()
        return {Module}Out.model_validate(row) if row else None

    async def create(
        self,
        request: Create{Module}Request,
        created_by: str,
    ) -> {Module}Out:
        """Create a new {module}."""
        # Implementation
        pass

schemas.py¶

"""{Module} Pydantic schemas."""
from datetime import datetime
from typing import Any
from pydantic import BaseModel, ConfigDict


class {Module}Base(BaseModel):
    """Base {module} fields."""
    name: str
    description: str | None = None


class {Module}Out({Module}Base):
    """Response schema for a {module}."""
    model_config = ConfigDict(from_attributes=True)

    id: str
    created_at: datetime
    updated_at: datetime
    created_by: str


class {Module}ListResponse(BaseModel):
    """Response schema for listing {modules}."""
    data: list[{Module}Out]
    meta: dict[str, Any]


class {Module}DetailResponse(BaseModel):
    """Response schema for a single {module}."""
    data: {Module}Out


class Create{Module}Request({Module}Base):
    """Request schema for creating a {module}."""
    pass

dependencies.py¶

"""{Module} FastAPI dependencies."""
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.database.postgres import get_session
from app.modules.{module}.repository import {Module}Repository
from app.modules.{module}.service import {Module}Service


async def get_{module}_repository(
    session: AsyncSession = Depends(get_session),
) -> {Module}Repository:
    return {Module}Repository(session)


async def get_{module}_service(
    repo: {Module}Repository = Depends(get_{module}_repository),
) -> {Module}Service:
    return {Module}Service(repo)

models.py (if using PostgreSQL)¶

"""{Module} SQLAlchemy models."""
from datetime import datetime
from sqlalchemy import String, Text, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column

from app.core.models.base import Base, TimestampMixin


class {Module}Model(Base, TimestampMixin):
    """{Module} database model."""
    __tablename__ = "{modules}"

    id: Mapped[str] = mapped_column(String(26), primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    description: Mapped[str | None] = mapped_column(Text)
    created_by: Mapped[str] = mapped_column(String(255), nullable=False)

init.py¶

"""{Module} package."""

Registering the Module¶

All feature modules must be registered in the central registry at backend/app/bootstrap/modules.py.

1. Update `backend/app/bootstrap/modules.py`¶

Import your router and add a new ApiModule entry to the API_MODULES tuple:

from app.modules.{module}.router import router as {module}_router

# ...

API_MODULES: tuple[ApiModule, ...] = (
    # ... existing modules
    ApiModule(
        name="{module}",
        router={module}_router,
        primary_tag="{modules}",
        bounded_context="{domain-context}",
        service_candidate="{service-name}",
        description="Detailed description of the module's responsibility.",
    ),
)

2. Metadata Requirements¶

When registering, you must provide: - name: Unique internal identifier. - router: The APIRouter instance from your module. - primary_tag: The primary OpenAPI tag for grouping in documentation. - bounded_context: The DDD bounded context this module belongs to. - service_candidate: The target microservice this module would extract into. - description: A clear summary of the module's functional scope.

Testing¶

Create tests in backend/tests/test_{module}.py:

import pytest
from httpx import AsyncClient


@pytest.mark.asyncio
async def test_list_{modules}(client: AsyncClient):
    response = await client.get("/api/v1/{modules}")
    assert response.status_code == 200
    data = response.json()
    assert "data" in data


@pytest.mark.asyncio
async def test_create_{module}(client: AsyncClient):
    response = await client.post(
        "/api/v1/{modules}",
        json={"name": "Test", "description": "Test description"},
    )
    assert response.status_code == 201

Best Practices¶

Use type hints throughout
Document functions with docstrings
Validate inputs with Pydantic schemas
Handle errors with domain exceptions
Use transactions in repositories
Test business logic in services
Mock dependencies in unit tests
Follow naming conventions (snake_case for files, PascalCase for classes)
Keep routers thin — business logic in services
Return envelopes — use list_response() and envelope_response()