# 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 `@`, 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 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. ## Related - [Projections](): the convergence model. - [Add a consumer](): views of other modules' channels. - [Events and consumers](): why there is no ordering. - [Write a migration]().