Skip to content
Menu

Documentation

Tutorial: your first feature

You build tasklist, a small multi-tenant application with two modules:

  • tasks stores task items and publishes tasks.item.created.v1 when one is created;
  • stats counts the items of each tenant by consuming that event exactly once, and provides a quota contract that tasks calls before creating an item.

Generators create the packages and boilerplate; you fill in complete domain files. All the code on this page comes from the finished application in examples/tutorial/, so you can compare your files with it at any point.

You need: Python 3.11 or later (3.12 recommended), uv 0.13.4+ and Docker with Compose. Start in an empty directory; no framework checkout is needed. It takes about twenty minutes.

1. Create the application

uvx --from "sandloom==0.7.0a6" sandloom new tasklist --no-batteries --module tasks
cd tasklist
uv sync
uv run sandloom generate
docker compose -f deploy/docker-compose.dev.yml up -d --wait   # Postgres and Redis

sandloom new writes a complete application: tasklist/asgi.py (the composition root), pyproject.toml with the test plugin activated, a deploy/ directory, an AGENTS.md for coding agents, and a first module, tasklist/modules/tasks/. The module's whole entry point is one line:

tasklist/modules/tasks/__init__.py
"""The tasks 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")

What Sandloom did for you: the application loads every subpackage of tasklist/modules that defines register, so a new module edits nothing else. define_module discovers the module's routes, events and tables.

2. Read the generated module

The table. TenantScoped gives it a tenant_id column, filled from the caller's tenant and filtered on every query:

tasklist/modules/tasks/models/item.py
"""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 TasksItem(TenantScoped, Base):
    """``TenantScoped``: rows belong to the bound tenant and are filtered by it."""

    __tablename__ = "tasks_items"

    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. The channel starts with the module's name, so this class is the channel's producer:

tasklist/modules/tasks/events.py
"""Events of the tasks module (its own channels)."""

from uuid import UUID

from sandloom.sdk import Event


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

The use cases. db() is the session of the running unit of work; the code flushes and never commits:

tasklist/modules/tasks/services.py
"""The tasks use cases. ``db()`` is the session of the running unit."""

from uuid import UUID

from sqlalchemy import select

from sandloom.sdk import db

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


async def create_item(body: ItemIn, *, owner_id: UUID) -> ItemOut:
    item = TasksItem(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(TasksItem, item_id)
    return None if item is None else ItemOut.model_validate(item)


async def list_items(*, limit: int) -> list[ItemOut]:
    newest = TasksItem.created_at.desc()
    tie = TasksItem.id.desc()
    query = select(TasksItem).order_by(newest, tie).limit(limit)
    rows = await db().scalars(query)
    return [ItemOut.model_validate(row) for row in rows]

What Sandloom did for you:

  • module_router() (in api.py) authenticates every route, binds the tenant and mounts the routes under /api/v1/tasks.
  • The write commits before the response is sent. The event commits in the same transaction (the outbox), so it is published if and only if the item is stored.
  • A retried POST with the same Idempotency-Key is not executed twice.
  • Before the write commits, the caller's session is checked again.

Nothing of this is declared in the code. See Route policy and sessions and Units of Work.

Run the module's tests and the gates:

uv run sandloom test tasks

3. Add a second module

uv run sandloom add module stats

The generator writes tasklist/modules/stats/ with an example table. Stats needs one counter row per tenant instead. Remove the unused example model and producer event:

rm tasklist/modules/stats/models/item.py tasklist/modules/stats/events.py

Create tasklist/modules/stats/models/counter.py with this complete content:

tasklist/modules/stats/models/counter.py
"""One counter row per tenant: how many task items it created."""

import uuid

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

from sandloom.sdk import Base, TenantScoped, new_id


class StatsItemCount(TenantScoped, Base):
    """``TenantScoped``: the row belongs to the bound tenant."""

    __tablename__ = "stats_item_counts"
    __table_args__ = (UniqueConstraint("tenant_id"),)

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=new_id)
    items: Mapped[int] = mapped_column(default=0)

Nothing has been deployed yet, so edit the generated baseline revision to create this table instead of the example one. Keep the row-level-security calls the generator wrote. The unique key on tenant_id already serves the tenant lookups, so the table gets no separate tenant_id index:

tasklist/modules/stats/migrations/versions/0001_stats_baseline.py
"""stats baseline: the item counters (rendered by `sandloom add module`, then edited
before the first deploy)

Revision ID: 0001_stats
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_stats"
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(
        "stats_item_counts",
        sa.Column("tenant_id", sa.Uuid(), nullable=False),
        sa.Column("id", sa.Uuid(), nullable=False),
        sa.Column("items", sa.Integer(), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("tenant_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, "stats_item_counts")


def downgrade() -> None:
    if op.get_context().config.attributes.get("sandloom_tenant_rls", False):
        disable_tenant_rls(op, "stats_item_counts")
    op.drop_table("stats_item_counts")

Once a revision has been applied anywhere, never edit it: add a new one with uv run sandloom migrate new stats "<message>" (Write a migration).

The use cases. The select needs no tenant filter: the bound tenant filters TenantScoped rows automatically.

tasklist/modules/stats/services.py
"""The stats use cases. ``db()`` is the session of the running unit."""

from sqlalchemy import select

from sandloom.sdk import db

from .models.counter import StatsItemCount

#: How many items one tenant may create (read by the quota contract).
ITEM_LIMIT = 100


async def _counter() -> StatsItemCount | None:
    # TenantScoped rows are filtered by the bound tenant: at most one row.
    result = await db().execute(select(StatsItemCount))
    return result.scalar_one_or_none()


async def record_item() -> None:
    counter = await _counter()
    if counter is None:
        db().add(StatsItemCount(items=1))
    else:
        counter.items += 1
    await db().flush()


async def item_count() -> int:
    counter = await _counter()
    return 0 if counter is None else counter.items

Stats consumes an event and produces none. Replace its example request/response schemas with this response DTO:

tasklist/modules/stats/schemas.py
"""Request and response shapes (pydantic only)."""

from pydantic import BaseModel


class CountOut(BaseModel):
    items: int

Replace its generated api.py with one read route:

tasklist/modules/stats/api.py
"""Routes of the stats module: ``/api/v1/stats/...``.

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

from sandloom.sdk import CurrentUser, module_router

from . import services
from .schemas import CountOut

router = module_router()


@router.get("/count")
async def get_count(user: CurrentUser) -> CountOut:
    return CountOut(items=await services.item_count())

The generated stats/tests/test_stats.py still tests the example CRUD routes. Replace its whole content with the tests in the next step before running tests. The counter is still on an unapplied baseline migration; no existing data is being changed.

4. Consume the event exactly once

uv run sandloom add consumer stats tasks.item.created.v1

The generator reads the producer's schema and writes an empty handler and a view of the event. A view is the stats module's own class for a channel it does not own. Make the handler count the item:

tasklist/modules/stats/consumers.py
"""Consumers of the stats module."""

from uuid import UUID

from sandloom.sdk import Event, on

from . import services


class ItemCreated(Event, channel="tasks.item.created.v1"):
    """A view: fields of the producer (tasks/events.py)."""

    item_id: UUID
    tenant_id: UUID


@on(ItemCreated)
async def on_item_created(event: ItemCreated) -> None:
    """Runs once per event (durable), with the event's tenant bound."""
    await services.record_item()

What Sandloom did for you:

  • the handler runs in its own unit of work, with the event's tenant_id bound, so services.record_item() writes the right tenant's row;
  • the consumer is durable: a redelivered event (same envelope id) is applied once;
  • stats never imports tasks: the view is checked against the producer's schema instead (Events).

The durable inbox deduplicates the same envelope during its configured retention window; separate event ids are separate occurrences (guarantees). Test this through the real consumer runtime. deliver(..., times=2) delivers the same event twice:

tasklist/modules/stats/tests/test_stats.py
"""Tests of the stats module (fixtures: the sandloom.sdk.testing plugin)."""

from typing import Any
from uuid import uuid4

from sandloom.sdk.testing import as_user, deliver

from ..consumers import ItemCreated


async def test_a_redelivered_event_is_counted_once(client: Any, db: Any) -> None:
    tenant = uuid4()
    _, headers = as_user(tenant_id=tenant)
    event = ItemCreated(item_id=uuid4(), tenant_id=tenant)
    await deliver(event, times=2, session=db)  # the same envelope id twice
    response = await client.get("/api/v1/stats/count", headers=headers)
    assert response.status_code == 200, response.text
    assert response.json() == {"items": 1}


async def test_counts_are_per_tenant(client: Any, db: Any) -> None:
    await deliver(ItemCreated(item_id=uuid4(), tenant_id=uuid4()), session=db)
    _, headers = as_user(tenant_id=uuid4())  # another tenant
    response = await client.get("/api/v1/stats/count", headers=headers)
    assert response.json() == {"items": 0}

Regenerate the files the new consumer changes, then run the stats tests:

uv run sandloom generate
uv run sandloom test stats

5. Provide a contract

Create tasklist/modules/stats/contracts.py with the final read-only provider below. Its public method returns a wire-safe bool, so it needs no extra DTO:

tasklist/modules/stats/contracts.py
"""Contracts the stats module provides."""

from sandloom.sdk import contract

from . import services


@contract("stats.quota.v1")
class Quota:
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step)."""

    async def may_create_item(self) -> bool:
        return await services.item_count() < services.ITEM_LIMIT

Generate the contract's first stub and snapshot:

uv run sandloom generate stats

We declare the final shape before its first generation. sandloom add contract is useful for a generic scaffold, but it immediately snapshots its example methods; removing those methods afterwards is a breaking change, even before publication. Once callers use a contract, preserve its methods or introduce a new version (contract evolution). Keep the generated files; sandloom test checks them without writing them.

The generated stub is:

tasklist/shared/contracts/stats_quota_v1.py
# Generated by `sandloom generate` (Public API 1.0): never edit by hand.
"""Contract ``stats.quota.v1``: the typed stub callers import.

Generated from ``tasklist.modules.stats.contracts.Quota``.
"""

from __future__ import annotations

from typing import Protocol

from sandloom.core.descriptors.contract_refs import ContractRef


class StatsQuotaV1(Protocol):
    """Contract ``stats.quota.v1``: public ``async def`` methods are reads,
    ``@command`` marks a command (deduplicated by its key, a saga step).
    """

    async def may_create_item(self) -> bool: ...


STATS_QUOTA_V1: ContractRef[StatsQuotaV1] = ContractRef("stats.quota.v1", StatsQuotaV1)

Callers import this stub, never the stats module.

6. Call the contract

Replace tasklist/modules/tasks/api.py with this complete file. The contract reference is at module level; the create route calls it before inserting an item, and the generated read/list routes remain available:

tasklist/modules/tasks/api.py
"""Routes of the tasks module: ``/api/v1/tasks/...``.

``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 Contract, CurrentUser, module_router
from tasklist.shared.contracts.stats_quota_v1 import STATS_QUOTA_V1

from . import services
from .schemas import ItemIn, ItemOut

# Required: items must never be created without the quota check.
quota = Contract(STATS_QUOTA_V1)
router = module_router()


@router.post("/items", status_code=201)
async def create_item(body: ItemIn, user: CurrentUser) -> ItemOut:
    if not await quota.may_create_item():
        raise HTTPException(status_code=402, detail="item quota reached")
    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; the limit caps each request."""
    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

What Sandloom did for you:

  • Contract(STATS_QUOTA_V1) is typed: your editor and mypy know may_create_item() returns a bool;
  • the module now declares that it consumes stats.quota.v1; a deployment without the provider fails at startup instead of creating items unchecked;
  • in a monolith the call runs in-process; when stats runs as its own service the same line calls it over internal RPC (Contracts).

Replace tasklist/modules/tasks/tests/test_tasks.py with this complete file. It includes the test double and imports needed to test both quota answers:

tasklist/modules/tasks/tests/test_tasks.py
"""Tests of the tasks module (fixtures: the sandloom.sdk.testing plugin)."""

from typing import Any
from uuid import uuid4

from sandloom.sdk.testing import as_user, fake_contract, published
from tasklist.shared.contracts.stats_quota_v1 import STATS_QUOTA_V1

from ..events import ItemCreated


class Quota:
    """A test double of ``stats.quota.v1``."""

    def __init__(self, allow: bool) -> None:
        self.allow = allow

    async def may_create_item(self) -> bool:
        return self.allow


async def test_create_and_read_an_item(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    with fake_contract(STATS_QUOTA_V1, Quota(allow=True)):
        created = await client.post(
            "/api/v1/tasks/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/tasks/items/{item['id']}", headers=headers)
    assert fetched.status_code == 200
    assert fetched.json()["title"] == "first"


async def test_the_quota_refuses_a_new_item(client: Any) -> None:
    _, headers = as_user(tenant_id=uuid4())
    with fake_contract(STATS_QUOTA_V1, Quota(allow=False)):
        response = await client.post(
            "/api/v1/tasks/items", json={"title": "x"}, headers=headers
        )
    assert response.status_code == 402
    assert published(ItemCreated) == []


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

The counter follows the outbox asynchronously. This read-only quota is a contract example, not an atomic hard limit under concurrent requests; a strict quota needs a reservation/command design (contracts).

7. Run every gate

uv run sandloom generate
uv run sandloom test --full

sandloom generate writes the contract stub, the event schemas and the module snapshots after a compatibility check; sandloom add already ran it after each generator. sandloom test --full runs the whole suite in parallel with ruff, format, mypy and sandloom check, and checks those files without writing them. Commit the generated files with your code.

Run the application:

uv run sandloom migrate upgrade-all --apply
uv run uvicorn tasklist.asgi:app --reload       # http://localhost:8000/docs

8. See where the infrastructure comes from

Your modules called db() and never named a database engine. The backend is chosen in .env:

DATABASE_URL=postgresql+asyncpg://...   # the database
CACHE_URL=redis://...                   # the cache
BROKER_URL=redis://...                  # the background task queue

Each setting configures a default driver, the piece of code that connects Sandloom to a backend. Pointing a setting at another server, or selecting another driver in INFRA, changes no module code. You have now used both main parts of Sandloom: modules hold your product, and drivers connect it to the outside world. See what is included and how to write a driver.

If you are stuck

  • Every error carries a code: uv run sandloom explain <code> says what happened and how to fix it. All codes: error catalogue.
  • uv run sandloom docs <name> prints a name's signature and documentation.
  • Compare your files with examples/tutorial/.

Next steps