Skip to content
Menu

Documentation

Add a projection

Goal: a table your module owns and keeps up to date from events, such as "boards with their title" or "members with their plan", so a list endpoint reads one table instead of joining other modules' data at request time.

Use it when an endpoint lists or filters data that other modules (or other aggregates of your module) own, and eventual consistency is acceptable.

Do not use it to check a strong invariant: a projection lags its sources, so call the owner's command contract and let it refuse. Do not use it for a single counter fed by one channel without versions; a plain consumer is enough. Background: Projections.

Prerequisites

  • The module exists and keeps its tables in a models/ package, as sandloom add module lays it out (the generator refuses otherwise).
  • The source events carry the aggregate id, tenant_id and a per-aggregate version that the producer increments on every change. Without a version a projection cannot order events, because delivery has no ordering guarantee.

Steps

1. Run the generator

sandloom add projection notes board

It writes three new files in the module package:

  • models/board_row.py: the row model;
  • projections.py: two example source events and the projection;
  • migrations/versions/notes_board_projection.py: the migration that creates the table, chained after the module chain's current head (the generator refuses a chain with several heads).

2. Read what it wrote

The row: one row per aggregate, TenantScoped, with the owned columns, the group's version column and the tombstone column:

"""Rows of the notes.board projection (a read model)."""

import uuid
from datetime import datetime

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

from sandloom.sdk import Base, TenantScoped


class BoardRow(TenantScoped, Base):
    __tablename__ = "notes_board_rows"

    board_id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
    title: Mapped[str | None] = mapped_column(String(255), nullable=True)
    version: Mapped[int | None] = mapped_column(Integer, nullable=True)
    deleted_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), nullable=True
    )

The projection:

"""The notes.board projection (spec §6.4).

Full-state events of one aggregate, versioned tombstones, one owning group
per column.
"""

from uuid import UUID

from sandloom.sdk import Event
from sandloom.sdk.projections import (  # sandloom: advanced projections
    Projection,
    SourceGroup,
    full_state,
    tombstone,
)

from .models.board_row import BoardRow


class BoardSaved(Event, channel="notes.board.saved.v1"):
    board_id: UUID
    tenant_id: UUID
    title: str
    version: int


class BoardDeleted(Event, channel="notes.board.deleted.v1"):
    board_id: UUID
    tenant_id: UUID
    version: int


BOARD = Projection(
    "notes.board",
    key=("tenant_id", "board_id"),
    model=BoardRow,
    groups=[
        SourceGroup(
            events={
                BoardSaved: full_state(lambda e: {"title": e.title}),
                BoardDeleted: tombstone(lambda e: {}),
            },
            aggregate_key="board_id",
            version_field="version",
            version_column="version",
            owns=("title",),
        )
    ],
)
  • key identifies a row and must start with tenant_id.
  • A SourceGroup is the set of channels of one aggregate (here saved and deleted) that share one version domain: aggregate_key and version_field name fields of every event in the group, version_column is the column that stores the version, and owns lists the columns this group writes. Every column is owned by exactly one group or seeded with insert_only.
  • Mappers turn an event into the owned fields:
  • full_state(fn): the event carries every owned field; the row is replaced only if the stored version is NULL or older. This converges for any interleaving and any number of duplicates.
  • tombstone(fn): a delete, stored as a versioned tombstone (deleted_at); reads never return it, and an older event cannot resurrect the row. A later full-state event clears it.
  • delta(fn): an increment ({"count": 1}), added to the stored value. Deltas are not convergent under "skip older", so they are allowed only in the consumer's own database under the durable inbox, which applies each event exactly once (not with at_least_once=True).
  • Discovery registers one durable consumer per source channel, named <projection>@<channel>, binding the tenant from the event's tenant_id. The projection's consumers apply each event in the consumer's unit of work.

The migration, with row-level security for the tenant table:

"""notes.board projection table (by `sandloom add projection`)

Revision ID: notes_board_projection
Revises: 0001_notes
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 = "notes_board_projection"
down_revision: str | Sequence[str] | None = "0001_notes"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None


def upgrade() -> None:
    op.create_table(
        "notes_board_rows",
        sa.Column("tenant_id", sa.Uuid(), nullable=False),
        sa.Column("board_id", sa.Uuid(), nullable=False),
        sa.Column("title", sa.String(length=255), nullable=True),
        sa.Column("version", sa.Integer(), nullable=True),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.PrimaryKeyConstraint("tenant_id", "board_id"),
    )
    op.create_index(op.f("ix_notes_board_rows_tenant_id"), "notes_board_rows", ["tenant_id"])
    if op.get_context().config.attributes.get("sandloom_tenant_rls", False):
        enable_tenant_rls(op, "notes_board_rows")


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

3. Point it at your real sources

The generated events are the module's own channels (notes.board.saved.v1), so they are producers defined next to the projection. In a real projection:

  1. Your own aggregate: move the event classes to events.py, import them in projections.py, and publish them from your services with the aggregate's version.
  2. Another module's aggregate: declare a view of its channel (an Event subclass with the fields you need, as sandloom add consumer writes it) and use it in the group. Never add a field the producer does not send.
  3. Several aggregates in one row (a member and their plan): one SourceGroup per aggregate, each with its own version column and its own owned columns.
  4. Change the model, then write a new revision with sandloom migrate new; edit the generated migration only while it has never been applied anywhere.

4. Read it

Inject a ProjectionView with View(name, row_type). Reads are filtered to the bound tenant and never return tombstoned rows:

from typing import Annotated
from sandloom.sdk.data import Page
from sandloom.sdk.projections import ProjectionView, View

@router.get("/boards")
async def list_boards(
    boards: Annotated[ProjectionView[BoardOut], View("notes.board", BoardOut)],
    cursor: str | None = None,
) -> Page[BoardOut]:
    return await boards.query(limit=50, cursor=cursor)

get(key) returns one row or None; query(filter=, limit=, cursor=) returns a page ordered by the key.

5. Make it rebuildable

A projection is a cache: rebuilding it from the sources is the repair path after corruption or a schema change. Pass rebuild_from= a read method of the source's contract, through its caller:

boards = Contract(NOTES_BOARDS_V1)

BOARD = Projection(..., rebuild_from=boards.list_for_tenant)

list_for_tenant(tenant_id) returns the aggregates' current states, as instances of the group's events or as mappings of its single full-state event. The rebuilder applies them through the same version-guarded path as the consumers and never infers a delete from a listing. Run it as a one-off job once the source services are up, never in the migrate one-shot:

sandloom projection list
sandloom projection rebuild notes.board --tenant <tenant id>
sandloom projection rebuild notes.board            # every tenant the projection holds

Purge tombstones

Tombstones must outlive every replay. Purge them only through the framework, never by hand:

sandloom projection purge notes.board --older-than-days 30

The purge refuses an age below the replay horizon (stream retention plus the relay's redelivery horizon, the oldest DLQ entry and the oldest dead outbox row of the source channels) and while any rebuild of the projection runs (each running rebuild is counted: one finishing never lifts the exclusion of another). It raises the projection's purge watermark before deleting, and from then on an event emitted before the watermark is dead-lettered instead of resurrecting a row.

Read your own writes

Set PROJECTION_TOKEN_KEYS ({kid: secret}). A write route whose events feed a projection then answers with SANDLOOM-Consistency-Token; the client sends it back on the read, and View(...) waits up to wait_ms for those rows before serving (stale reads carry SANDLOOM-Projection-Stale: true).

ProjectionView.search() needs a store with the search feature; none ships in 1.0, so it raises NotImplementedError.

Verify

Deliver source events in a test and read the projection back: deliver an older version after a newer one and check the newer fields win; deliver a tombstone and check the row disappears from query(); deliver the same event twice (deliver(event, times=2)).

sandloom test notes

Errors you may hit

  • SANDLOOM-B012: the key does not start with tenant_id; a column is owned by two groups, or is both owned and insert_only; two groups share a version column; a mapper is not full_state, delta or tombstone; a delta runs without the inbox; a tombstone has no deleted_at column; an event lacks the group's aggregate_key or version_field; the model lacks a declared column.
  • SANDLOOM-B006: the projection is defined outside the module's projections submodule.
  • SANDLOOM-B015: a source event class is invalid.
  • SANDLOOM-T002: the row table is not created by the module's migration chain.