# Add a module

**Goal:** a new business module (a package under `<app>/modules/`) with a table, an authenticated route, an event, a migration and tests, loaded by the app with no wiring.

**Use it** for a new bounded part of your domain that owns its own tables, such as `billing_notes`, `stats` or `board`.

**Do not use it** to split code inside one module (add a file to the existing package instead), or to share code between modules. Modules never import each other: they talk through [events](<https://sandloom.dev/docs/modules/between/events/index.md>) and [contracts](<https://sandloom.dev/docs/modules/between/contracts/index.md>). See [Modules](<https://sandloom.dev/docs/modules/index.md>).

## Prerequisites

- An application created with `sandloom new` (it has an `architecture_rules.yml` at its root).
- The development services running: `docker compose -f deploy/docker-compose.dev.yml up -d --wait`.

## Steps

### 1\. Run the generator

From the application root:

```
sandloom add module board
```

It writes new files inside `<app>/modules/board/` only, and refuses to run if the package already exists. It edits no other file: the app loads every subpackage of `<app>/modules` whose `__init__.py` defines `register`, so the new module is part of the app at once. `SANDLOOM_MODULES` still selects a subset in a given process.

### 2\. Read what it wrote

The module declaration. `define_module` discovers everything else from the conventional submodules (`models`, `api`, `events`, `consumers`, `contracts`, `sagas`, `projections`, `tasks`):

```
"""The board module.

Everything it declares is discovered: routes in ``api.py``, events in
``events.py``, tables in ``models/``. It talks to other modules only through
events and contracts.
"""

from sandloom.sdk import define_module

register = define_module(__name__, version="0.1.0")
```

An example table. `TenantScoped` adds the `tenant_id` column and filters every query by the bound tenant. Table names start with the module name. `created_at: Mapped[CreatedAt]` defaults to `clock_timestamp()`, not `now()`: `now()` is the transaction's start, so rows created in one transaction (a test's `client` fixture shares one) would tie, and a "newest first" list would come back in any order. `default=new_id` gives time-ordered UUIDv7 ids. The list orders by `created_at`, then `id` ([Timestamps, ids and list order](<https://sandloom.dev/docs/modules/database/conventions/index.md>)):

```
"""An example table: replace it with your domain."""

import uuid

from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column

from sandloom.sdk import Base, CreatedAt, TenantScoped, new_id


class BoardItem(TenantScoped, Base):
    """``TenantScoped``: rows belong to the bound tenant and are filtered by it."""

    __tablename__ = "board_items"

    # new_id: a time-ordered UUIDv7; CreatedAt: clock_timestamp(), not now()
    # (the transaction's start), so rows of one transaction never tie.
    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=new_id)
    title: Mapped[str] = mapped_column(String(255))
    owner_id: Mapped[uuid.UUID] = mapped_column(index=True)
    created_at: Mapped[CreatedAt]
```

The event the module publishes. Its channel starts with the module name, so this class is the channel's producer:

```
"""Events of the board module (its own channels)."""

from uuid import UUID

from sandloom.sdk import Event


class ItemCreated(Event, channel="board.item.created.v1"):
    item_id: UUID
    tenant_id: UUID
```

The use cases. `db()` is the session of the running unit of work; the event is written to the outbox in the same transaction as the row:

```
"""The board use cases. ``db()`` is the session of the running unit."""

from uuid import UUID

import sqlalchemy as sa

from sandloom.sdk import db

from .events import ItemCreated
from .models.item import BoardItem
from .schemas import ItemIn, ItemOut


async def create_item(body: ItemIn, *, owner_id: UUID) -> ItemOut:
    item = BoardItem(title=body.title, owner_id=owner_id)
    db().add(item)
    await db().flush()
    await ItemCreated(item_id=item.id, tenant_id=item.tenant_id).publish()
    return ItemOut.model_validate(item)


async def get_item(item_id: UUID) -> ItemOut | None:
    item = await db().get(BoardItem, item_id)
    return None if item is None else ItemOut.model_validate(item)


async def list_items(*, limit: int) -> list[ItemOut]:
    """The newest items first: ``created_at`` (the database clock), then the
    UUIDv7 ``id`` (creation order) for rows stamped in the same microsecond.
    ``TenantScoped`` keeps them to the bound tenant."""
    newest = BoardItem.created_at.desc()
    tie = BoardItem.id.desc()
    query = sa.select(BoardItem).order_by(newest, tie).limit(limit)
    rows = await db().scalars(query)
    return [ItemOut.model_validate(row) for row in rows]
```

The routes. `module_router()` mounts them under `/api/v1/board`, authenticates every route and binds the tenant (the module has a `TenantScoped` table). Nothing else is declared: the `POST` gets the write defaults (automatic idempotency, the commit-time session check). The list route caps its `limit` with `Query(le=...)`: one request never loads an unbounded table, and `TenantScoped` keeps it to the caller's tenant:

```
"""Routes of the board module: ``/api/v1/board/...``.

``module_router()`` authenticates every route and binds the tenant.
"""

from typing import Annotated
from uuid import UUID

from fastapi import HTTPException, Query

from sandloom.sdk import CurrentUser, module_router

from . import services
from .schemas import ItemIn, ItemOut

router = module_router()


@router.post("/items", status_code=201)
async def create_item(body: ItemIn, user: CurrentUser) -> ItemOut:
    return await services.create_item(body, owner_id=user.user_id)


@router.get("/items")
async def list_items(
    user: CurrentUser, limit: Annotated[int, Query(ge=1, le=100)] = 20
) -> list[ItemOut]:
    """A page of the tenant's items: ``le=`` caps what one request may load."""
    return await services.list_items(limit=limit)


@router.get("/items/{item_id}")
async def get_item(item_id: UUID, user: CurrentUser) -> ItemOut:
    item = await services.get_item(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="item not found")
    return item
```

The baseline migration of the module's own chain, rendered from the model (no database is needed). It enables row-level security for the tenant table:

```
"""board baseline: the example table (rendered by `sandloom add module`)

Revision ID: 0001_board
Revises:
Create Date: generated

"""

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

from sandloom.sdk.data import disable_tenant_rls, enable_tenant_rls  # sandloom: advanced RLS

revision: str = "0001_board"
down_revision: str | Sequence[str] | None = None
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    op.create_table(
        "board_items",
        sa.Column("tenant_id", sa.Uuid(), nullable=False),
        sa.Column("id", sa.Uuid(), nullable=False),
        sa.Column("title", sa.String(length=255), nullable=False),
        sa.Column("owner_id", sa.Uuid(), nullable=False),
        sa.Column(
            "created_at",
            sa.DateTime(timezone=True),
            server_default=sa.text("clock_timestamp()"),
            nullable=False,
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_board_items_tenant_id"), "board_items", ["tenant_id"])
    op.create_index(op.f("ix_board_items_owner_id"), "board_items", ["owner_id"])
    # The runner sets sandloom_tenant_rls from the target database's INFRA config.
    if op.get_context().config.attributes.get("sandloom_tenant_rls", False):
        enable_tenant_rls(op, "board_items")


def downgrade() -> None:
    if op.get_context().config.attributes.get("sandloom_tenant_rls", False):
        disable_tenant_rls(op, "board_items")
    op.drop_index(op.f("ix_board_items_owner_id"), table_name="board_items")
    op.drop_index(op.f("ix_board_items_tenant_id"), table_name="board_items")
    op.drop_table("board_items")
```

The tests, which run against the real app and a real database:

```
"""Tests of the board module (sandloom.sdk.testing fixtures)."""

from typing import Any
from uuid import uuid4

from sandloom.sdk.testing import as_user, published

from ..events import ItemCreated


async def test_create_and_read_an_item(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    created = await client.post(
        "/api/v1/board/items", json={"title": "first"}, headers=headers
    )
    assert created.status_code == 201, created.text
    assert len(published(ItemCreated)) == 1
    item = created.json()
    fetched = await client.get(f"/api/v1/board/items/{item['id']}", headers=headers)
    assert fetched.status_code == 200
    assert fetched.json()["title"] == "first"


async def test_list_items_is_capped_and_tenant_scoped(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    _, other = as_user(tenant_id=uuid4())
    for title in ("a", "b", "c"):
        created = await client.post(
            "/api/v1/board/items", json={"title": title}, headers=headers
        )
        assert created.status_code == 201, created.text
    listed = await client.get("/api/v1/board/items", headers=headers)
    assert listed.status_code == 200
    # Newest first, deterministically: all three rows share one transaction.
    assert [item["title"] for item in listed.json()] == ["c", "b", "a"]
    page = await client.get("/api/v1/board/items", params={"limit": 2}, headers=headers)
    assert len(page.json()) == 2
    foreign = await client.get("/api/v1/board/items", headers=other)
    assert foreign.json() == []
    over_cap = await client.get(
        "/api/v1/board/items", params={"limit": 101}, headers=headers
    )
    assert over_cap.status_code == 422


async def test_items_need_an_authenticated_user(client: Any) -> None:
    response = await client.post("/api/v1/board/items", json={"title": "x"})
    assert response.status_code == 401
```

The generator also writes `schemas.py` (pydantic request and response models), `models/__init__.py` (imports every `models/*.py`), `migrations/env.py` and `migrations/script.py.mako`.

### 3\. Replace the example with your domain

1. Rename or replace `BoardItem` in `models/`. Add one file per table; keep the `board_` prefix and reference other modules' rows by UUID, never by a foreign key.
2. If the module has not been deployed anywhere yet, edit the baseline migration so it creates your real tables. After the first deployment, never edit a revision that ran: add a new one ([Write a migration](<https://sandloom.dev/docs/modules/database/write-migration/index.md>)).
3. Adapt `schemas.py`, `services.py` and `api.py`. Keep SQL in `services.py` and `models/`, endpoints in `api.py`, and pure pydantic in `schemas.py`.
4. Rename the event or add more with [`sandloom add event`](<https://sandloom.dev/docs/modules/between/add-event/index.md>).

## Verify

```
sandloom test board
```

It runs the module's tests first, then the fast suite and the static gates (ruff, format, mypy, `sandloom check`, the route and contract snapshots). It never writes the module's generated files, such as `routes.snapshot.json`: `sandloom add` ran `sandloom generate`, which wrote them; after a change run `sandloom generate`, review them in your diff and commit them. A stale one fails with "generated files are stale: run `sandloom generate`".

Then apply the migration to your development database:

```
sandloom migrate upgrade-all --apply
```

## Errors you may hit

- [SANDLOOM-C001](<https://sandloom.dev/docs/reference/error-codes/#sandloom-c001>): the module imports another module; use an event or a contract.
- [SANDLOOM-B001](<https://sandloom.dev/docs/reference/error-codes/#sandloom-b001>): the module failed to load; the message carries the original traceback.
- [SANDLOOM-B006](<https://sandloom.dev/docs/reference/error-codes/#sandloom-b006>): an object `define_module` collects lives outside the conventional submodules.
- [SANDLOOM-B008](<https://sandloom.dev/docs/reference/error-codes/#sandloom-b008>): a route is neither authenticated nor `public(reason=...)`, or is both.
- [SANDLOOM-T001](<https://sandloom.dev/docs/reference/error-codes/#sandloom-t001>): the route snapshot is stale (run `sandloom generate`), or a route lost a protection (review it, then approve it with `sandloom routes snapshot --write`).
- [SANDLOOM-T002](<https://sandloom.dev/docs/reference/error-codes/#sandloom-t002>): a table exists in the models but no revision of the chain creates it.

## Related

- [Modules](<https://sandloom.dev/docs/modules/index.md>): the isolation rule and discovery.
- [Units of Work and `db()`](<https://sandloom.dev/docs/modules/database/units/index.md>).
- [Route policy](<https://sandloom.dev/docs/modules/http/route-policy/index.md>): the defaults every route gets.
- [Testing modules](<https://sandloom.dev/docs/toolchain/testing/index.md>).
- [Tutorial](<https://sandloom.dev/docs/getting-started/tutorial/index.md>): a module built step by step.