Skip to content
Menu

Documentation

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 and contracts. See Modules.

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):

"""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).
  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.

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: the module imports another module; use an event or a contract.
  • SANDLOOM-B001: the module failed to load; the message carries the original traceback.
  • SANDLOOM-B006: an object define_module collects lives outside the conventional submodules.
  • SANDLOOM-B008: a route is neither authenticated nor public(reason=...), or is both.
  • 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: a table exists in the models but no revision of the chain creates it.