# `sandloom.sdk.projections`

Query projections: the declarative `Projection` and today's descriptors and helpers.

Public API 1.0 (spec §2.2, Appendix C). A name whose mechanism lands at a later hand-off raises `NotImplementedError` naming it.

`from sandloom.sdk.projections import <name>`. Signatures in a terminal: `sandloom docs <name>`. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line `# sandloom: advanced <reason>` in generated code and examples).

## Advanced

| Name | Kind | Summary |
| --- | --- | --- |
| [`BoundContractMethod`](<#boundcontractmethod>) | callable |  |
| [`ConditionalWriteStore`](<#conditionalwritestore>) | class | Driver feature `conditional_write` (spec §2.3). |
| [`ConsistencyToken`](<#consistencytoken>) | class | A verified token (spec §2.3). |
| [`delta`](<#delta>) | function | A mapper of delta events (column increments); allowed only for a relational projection applied under the durable inbox (§6.4 item 4). |
| [`full_state`](<#full_state>) | function | A mapper of full-state events: every owned field, replaced iff the stored version is NULL or older (convergent under any interleaving). |
| [`make_consumer_callback`](<#make_consumer_callback>) | function | Wrap a projection consumer descriptor into an event-bus callback. |
| [`Mapper`](<#mapper>) | class | Maps an event to the projection fields its group owns. |
| [`monotonic_delete`](<#monotonic_delete>) | async function | Persist a versioned tombstone, including when deletion arrives first. |
| [`monotonic_update`](<#monotonic_update>) | async function | Update every projection row matching `match`, guarded by source version. |
| [`monotonic_upsert`](<#monotonic_upsert>) | async function | Insert-or-update a single projection row, guarded by source version. |
| [`Projection`](<#projection>) | class | A declared read model (spec §6.4), registered by `define_module` from the module's `projections` submodule. |
| [`ProjectionConsumerDescriptor`](<#projectionconsumerdescriptor>) | class | A durable consumer that applies one source channel to one projection. |
| [`ProjectionHandler`](<#projectionhandler>) | callable |  |
| [`ProjectionRebuilder`](<#projectionrebuilder_1>) | class | Reconstructs a projection from its source modules (idempotently). |
| [`ProjectionRebuildResult`](<#projectionrebuildresult>) | class | Outcome of a rebuild run (rows written + whether the index was cleared). |
| [`ProjectionRegistryError`](<#projectionregistryerror>) | exception | A projection/consumer registration conflicts with existing state. |
| [`ProjectionView`](<#projectionview>) | class | Reads one projection (`Annotated[ProjectionView[R], View(...)]`). |
| [`purge_tombstones`](<#purge_tombstones>) | async function | Purge tombstones behind the watermark that refuses older replays. |
| [`QueryProjectionDescriptor`](<#queryprojectiondescriptor>) | class | A denormalised cross-module read model owned by one module. |
| [`QueryProjectionRegistry`](<#queryprojectionregistry>) | class | Holds query-projection descriptors, consumers and rebuilders by name. |
| [`read_token`](<#read_token>) | function | The token of `header`; `None` for a missing, malformed, expired, unknown-kid or other-tenant token (never an error). |
| [`rebuild`](<#rebuild>) | async function | Rebuild projection `name` from its source (§6.4): one tenant, or every tenant. Purges refuse while it runs; the purge watermark is kept.... |
| [`SourceGroup`](<#sourcegroup>) | class | The channels of ONE aggregate feeding a projection, with its version. |
| [`start_projection_consumers`](<#start_projection_consumers>) | async function | Subscribe every registered projection consumer to the event bus. |
| [`tombstone`](<#tombstone>) | function | A mapper of delete events: a versioned tombstone in the owning group (§6.4 item 5); reads filter it. |
| [`View`](<#view>) | function | The dependency resolving a `ProjectionView` of projection `name`: |

## Advanced names

### `BoundContractMethod`

```
def BoundContractMethod(*args, **kwargs)
```

*(no docstring)*

### `ConditionalWriteStore`

```
class ConditionalWriteStore(Protocol)
```

Driver feature `conditional_write` (spec §2.3).

#### `ConditionalWriteStore.put_if_newer`

```
async def put_if_newer(
    self,
    *,
    tenant_id: uuid.UUID,
    key: Mapping[str, Any],
    version_column: str,
    version: int,
    fields: Mapping[str, Any],
    insert_only: Mapping[str, Any] | None = None,
) -> bool
```

#### `ConditionalWriteStore.tombstone_if_newer`

```
async def tombstone_if_newer(
    self,
    *,
    tenant_id: uuid.UUID,
    key: Mapping[str, Any],
    version_column: str,
    version: int,
    deleted_at: datetime,
) -> bool
```

### `ConsistencyToken`

```
class ConsistencyToken(
    kid: str,
    issued_at: int,
    tenant_id: UUID,
    entries: tuple[TokenEntry, ...],
) -> None
```

A verified token (spec §2.3).

#### `ConsistencyToken.digest`

```
@staticmethod
def digest(key: Mapping[str, Any]) -> str
```

The digest of a projection row key, as the entries carry it.

#### `ConsistencyToken.versions`

```
def versions(self, projection: str) -> dict[EntryKey, int]
```

`(key digest, group, aggregate digest) -> version` the token waits for in `projection`: one version per source group and aggregate, never merged across groups.

#### `ConsistencyToken.keys`

```
def keys(self, projection: str) -> set[str]
```

The row key digests the token names in `projection`.

#### `ConsistencyToken.wait_for`

```
async def wait_for(
    self,
    projection: str,
    stored_version: Callable[[str, str], Awaitable[int | None]],
    *,
    wait_ms: int | None = None,
    refresh: Callable[[], Awaitable[None]] | None = None,
) -> bool
```

Poll `stored_version(digest, group)` (the row's value of that group's version column) until `projection` reached every entry, at most `wait_ms` (default `PROJECTION_RYW_WAIT_MS`); `False` when the budget ran out (serve stale, never an error). `refresh` runs before every round but the first (a caller polling cached rows rereads them there).

### `delta`

```
def delta(fn: Callable[[E], Mapping[str, Any]]) -> Mapper[E]
```

A mapper of delta events (column increments); allowed only for a relational projection applied under the durable inbox (§6.4 item 4).

### `full_state`

```
def full_state(fn: Callable[[E], Mapping[str, Any]]) -> Mapper[E]
```

A mapper of full-state events: every owned field, replaced iff the stored version is NULL or older (convergent under any interleaving).

### `make_consumer_callback`

```
def make_consumer_callback(
    descriptor: ProjectionConsumerDescriptor,
    storage: str,
    opener: SessionOpener,
) -> Any
```

Wrap a projection consumer descriptor into an event-bus callback.

Delegates to the generic consumer wrapper with projection labels/metrics. Exposed separately so it can be unit-tested without a live bus.

### `Mapper`

```
class Mapper(Protocol)
```

Maps an event to the projection fields its group owns.

### `monotonic_delete`

```
async def monotonic_delete(
    session: AsyncSession,
    target: Any,
    *,
    key: dict[str, Any],
    version_column: str,
    incoming_version: int,
    values: dict[str, Any] | None = None,
    insert_only: dict[str, Any] | None = None,
    deleted_column: str = 'deleted_at',
    deleted_at: datetime | None = None,
) -> bool
```

Persist a versioned tombstone, including when deletion arrives first.

A physical DELETE discards the version and lets a delayed older upsert resurrect the row. This helper retains it using the same tenant/version checks as `monotonic_upsert`. The projection needs a nullable timestamp `deleted_column` and reads must filter it to NULL. `values` and `insert_only` must supply tenant\_id and other required columns so an absent row can be inserted as a tombstone. Only an explicitly newer event can subsequently clear it by upserting `deleted_column=None`.

Tombstones must survive at least the complete replay/rebuild horizon; removing them early reintroduces resurrection. Rebuilders apply source versions through the same helper and must not infer deletion from an incomplete source listing.

### `monotonic_update`

```
async def monotonic_update(
    session: AsyncSession,
    target: Any,
    *,
    match: dict[str, Any],
    values: dict[str, Any],
    version_column: str,
    incoming_version: int,
) -> int
```

Update every projection row matching `match`, guarded by source version.

For denormalised fields that fan out across many rows (e.g. a project's name copied onto every task row): updates the matched rows whose stored `version_column` is strictly older than `incoming_version` (or NULL), setting the new `values` and version. Out-of-order/duplicate project events are no-ops. Returns the number of rows updated. Auto-scoped to the bound tenant (if any) on tenant-owned tables — see the module docstring.

Note: this only refreshes rows that ALREADY EXIST and are COMMITTED (or locked by this transaction). A row another writer inserted but has not committed yet is invisible here, so a writer that reads the source over RPC and then inserts can lose this update for good. Serialise the two writers on a local, version-guarded snapshot row of the source (upsert it here first; `SELECT ... FOR UPDATE` it in the other writer and copy its values) — see the projecthub taskboard `ProjectSnapshot`.

### `monotonic_upsert`

```
async def monotonic_upsert(
    session: AsyncSession,
    target: Any,
    *,
    key: dict[str, Any],
    values: dict[str, Any],
    version_column: str,
    incoming_version: int,
    insert_only: dict[str, Any] | None = None,
) -> bool
```

Insert-or-update a single projection row, guarded by source version.

Inserts the row (`key` + `values` + `insert_only` + the version) when absent; on a primary-key/unique conflict it updates `values` and the version ONLY when the stored `version_column` is strictly older than `incoming_version` (or NULL). A duplicate (equal version) or an out-of-order older event leaves the row untouched. Returns `True` when a row was inserted or updated.

`insert_only` columns are written on INSERT but NEVER touched by the on-conflict update — for fields this source merely SEEDS but does not own (e.g. a task upsert seeds the denormalised project name once, but project events, guarded by their own version, are the authority for keeping it fresh). Without this, every task update would clobber project fields back to whatever the seed lookup returned, rolling back fresher project data.

### `Projection`

```
class Projection(
    name: str,
    *,
    key: tuple[str, ...],
    groups: Sequence[SourceGroup],
    model: type[TenantScoped] | None = None,
    store: tuple[str, str] | None = None,
    insert_only: Mapping[str, Any] | None = None,
    deleted_column: str = 'deleted_at',
    rebuild_from: BoundContractMethod | tuple[str, str] | None = None,
    lag_slo_s: float = 5.0,
    at_least_once: bool = False,
) -> None
```

A declared read model (spec §6.4), registered by `define_module` from the module's `projections` submodule.

`model=` (a relational `TenantScoped` table) is the supported target in 1.0. `store=` (a `document_store` instance) is experimental in 1.0: tested against the in-memory driver only, without production drivers.

#### `Projection.channels`

```
def channels(self) -> tuple[str, ...]
```

The source channels, in declaration order.

#### `Projection.descriptor`

```
def descriptor(self, owner_module: str) -> QueryProjectionDescriptor
```

#### `Projection.token_entries`

```
def token_entries(
    self,
    channel: str,
    payload: Mapping[str, Any],
) -> Iterable[tuple[Mapping[str, Any], str, Any, int]]
```

The `(row key, group, aggregate id, version)` an event of `channel` will produce (`group`: its source group's version column): a write's consistency token (spec §6.4 item 7).

#### `Projection.target`

```
def target(self) -> ConditionalWriteStore
```

The store of a `store=` projection (`conditional_write`).

#### `Projection.consumers`

```
def consumers(self) -> list[ProjectionConsumerDescriptor]
```

#### `Projection.rebuilder`

```
def rebuilder(self) -> ProjectionRebuilderFromContract | None
```

#### `Projection.apply`

```
async def apply(
    self,
    session: Any,
    group: SourceGroup,
    event: Event,
    mapper: Mapper[Any],
) -> bool
```

Apply one event to the projection in `session` (the consumer's unit).

### `ProjectionConsumerDescriptor`

```
class ProjectionConsumerDescriptor(
    name: str,
    source_channel: str,
    projection: str,
    handler: ProjectionHandler,
    durable: bool = True,
    required: bool = True,
    module: str | None = None,
    group: str | None = None,
    start_from: str = 'earliest',
    bind_tenant_from: str | None = None,
) -> None
```

A durable consumer that applies one source channel to one projection.

The durable consumer-group id on the event bus is `group_name` (`<module>:<name>` when registered through `ModuleAPI`; the module is stamped by the loader): each consumer gets its OWN group, so it has its own pending list, redelivery and dead-letter queue, independent of every other subscriber to the channel. Keep `name` stable across deploys — changing it makes the bus treat the consumer as brand new (a new group starting at `start_from`, default the channel's retained backlog). `group` pins an explicit group id instead.

`durable` (default) claims each event id in the handler's Unit of Work (`UNIQUE(consumer, event_id)`) so the apply runs exactly once even though delivery is at-least-once; idempotent handlers may set it `False` and rely on the monotonic upsert alone.

`required=True` makes the projection subscription part of readiness. Use `False` only when the owning read path explicitly tolerates a stale or unavailable projection.

#### `ProjectionConsumerDescriptor.group_name`

```
group_name  (property)
```

The durable consumer-group id (also the exactly-once inbox key).

### `ProjectionHandler`

```
def ProjectionHandler(*args, **kwargs)
```

*(no docstring)*

### `ProjectionRebuilder`

```
class ProjectionRebuilder(Protocol)
```

Reconstructs a projection from its source modules (idempotently).

Every method rebuilds a SCOPE and returns a `ProjectionRebuildResult`. Implementations re-read source state via public contracts/events and apply it through the same monotonic upsert the live consumers use, so a rebuild and steady-state delivery converge to the same rows. Rebuilds must be safe to run while consumers are live. Source deletions require versioned tombstones (`monotonic_delete`), never physical DELETEs: a stale source row observed before a concurrent delete must not resurrect a row during the rebuild. Retain tombstones through the whole replay/rebuild horizon and distinguish complete snapshots from partial listings before reconciling absent entities.

#### `ProjectionRebuilder.rebuild_all`

```
async def rebuild_all(self) -> ProjectionRebuildResult
```

Rebuild the whole projection (every tenant).

#### `ProjectionRebuilder.rebuild_tenant`

```
async def rebuild_tenant(self, tenant_id: uuid.UUID) -> ProjectionRebuildResult
```

Rebuild only the rows belonging to `tenant_id`.

#### `ProjectionRebuilder.rebuild_entity`

```
async def rebuild_entity(self, ref: str) -> ProjectionRebuildResult
```

Rebuild a single entity (a stable `owner_module:entity_id` ref).

### `ProjectionRebuildResult`

```
class ProjectionRebuildResult(
    projection: str,
    rows: int = 0,
    cleared: bool = False,
    details: dict[str, Any] = <factory>,
) -> None
```

Outcome of a rebuild run (rows written + whether the index was cleared).

### `ProjectionRegistryError`

```
class ProjectionRegistryError(RuntimeError)
```

A projection/consumer registration conflicts with existing state.

### `ProjectionView`

```
class ProjectionView(Generic)(
    projection: Projection,
    row: type[R],
    session: Any,
    *,
    on_stale: Literal['serve', 'source'] = 'serve',
    wait_ms: int | None = None,
    token: ConsistencyToken | None = None,
    response: Response | None = None,
) -> None
```

Reads one projection (`Annotated[ProjectionView[R], View(...)]`).

Rows whose tombstone is set are never returned. With a consistency token (`SANDLOOM-Consistency-Token`) a read waits up to `wait_ms` for the rows the caller's write produces; then it serves anyway with `stale = True` and `SANDLOOM-Projection-Stale: true` (`on_stale="serve"`), or reads the entity from its source through `rebuild_from` (`on_stale="source"`, `get` only). Never an error status.

#### `ProjectionView.get`

```
async def get(self, key: Mapping[str, Any]) -> R | None
```

#### `ProjectionView.query`

```
async def query(
    self,
    *,
    filter: Mapping[str, Any] | None = None,
    limit: int = 50,
    cursor: str | None = None,
) -> Page[R]
```

#### `ProjectionView.search`

```
async def search(self, q: str, *, limit: int = 50) -> list[R]
```

### `purge_tombstones`

```
async def purge_tombstones(name: str, *, older_than: timedelta) -> int
```

Purge tombstones behind the watermark that refuses older replays.

### `QueryProjectionDescriptor`

```
class QueryProjectionDescriptor(
    name: str,
    owner_module: str,
    version: str = 'v1',
    source_events: tuple[str, ...] = (),
    storage: str = 'default',
    consistency: Literal['eventual'] = 'eventual',
    rebuildable: bool = True,
    purge: TombstonePurger | None = None,
    token_entries: TokenEntries | None = None,
) -> None
```

A denormalised cross-module read model owned by one module.

`consistency` is intentionally fixed to `"eventual"`: a projection lags its source modules by the event delivery + apply time and must never back a strong-consistency command invariant (validate those against the owner's public command contract instead). `rebuildable` records that the projection can be dropped and reconstructed from the source modules — the projection is a cache, not a source of truth.

### `QueryProjectionRegistry`

```
class QueryProjectionRegistry() -> None
```

Holds query-projection descriptors, consumers and rebuilders by name.

#### `QueryProjectionRegistry.register_projection`

```
def register_projection(self, descriptor: QueryProjectionDescriptor) -> None
```

#### `QueryProjectionRegistry.register_consumer`

```
def register_consumer(self, descriptor: ProjectionConsumerDescriptor) -> None
```

#### `QueryProjectionRegistry.register_rebuilder`

```
def register_rebuilder(self, projection: str, rebuilder: ProjectionRebuilder) -> None
```

#### `QueryProjectionRegistry.get_projection`

```
def get_projection(self, name: str) -> QueryProjectionDescriptor | None
```

#### `QueryProjectionRegistry.get_consumer`

```
def get_consumer(self, name: str) -> ProjectionConsumerDescriptor | None
```

Look a consumer up by its group id, or by its bare name.

#### `QueryProjectionRegistry.get_rebuilder`

```
def get_rebuilder(self, projection: str) -> ProjectionRebuilder | None
```

#### `QueryProjectionRegistry.list_projections`

```
def list_projections(self) -> list[QueryProjectionDescriptor]
```

#### `QueryProjectionRegistry.list_consumers`

```
def list_consumers(self) -> list[ProjectionConsumerDescriptor]
```

#### `QueryProjectionRegistry.consumers_for`

```
def consumers_for(self, projection: str) -> list[ProjectionConsumerDescriptor]
```

#### `QueryProjectionRegistry.token_sources`

```
def token_sources(self) -> dict[str, tuple[tuple[str, Any], ...]]
```

`channel -> ((projection, token_entries), ...)` for every projection that issues consistency tokens (empty: none does).

#### `QueryProjectionRegistry.snapshot`

```
def snapshot(self) -> dict[str, dict[str, object]]
```

Shallow copy of the maps for a later `restore`.

Descriptors are frozen/immutable and rebuilders are opaque objects, so a shallow per-map copy is enough to rewind a failed module registration without leaking later mutations into the snapshot.

#### `QueryProjectionRegistry.restore`

```
def restore(self, snapshot: dict[str, dict[str, object]]) -> None
```

Reset the maps to a prior `snapshot`.

### `read_token`

```
def read_token(header: str | None) -> ConsistencyToken | None
```

The token of `header`; `None` for a missing, malformed, expired, unknown-kid or other-tenant token (never an error).

### `rebuild`

```
async def rebuild(
    name: str,
    *,
    tenant_id: UUID | None = None,
    blue_green: bool = True,
) -> ProjectionRebuildResult
```

Rebuild projection `name` from its source (§6.4): one tenant, or every tenant. Purges refuse while it runs; the purge watermark is kept. Without an alias-swapping store a blue/green rebuild runs in place (`details["blue_green"] == "in_place"`).

### `SourceGroup`

```
class SourceGroup(
    events: Mapping[type[Event], Mapper[Any]],
    aggregate_key: str,
    version_field: str,
    version_column: str,
    owns: tuple[str, ...],
    tenant_from: str = 'tenant_id',
) -> None
```

The channels of ONE aggregate feeding a projection, with its version.

### `start_projection_consumers`

```
async def start_projection_consumers(
    core: CoreFacade,
    *,
    session_opener: SessionOpener | None = None,
) -> int
```

Subscribe every registered projection consumer to the event bus.

Returns the number of consumers wired through the canonical supervised runtime. Duplicate groups and missing required projections fail startup; repeated calls do not add duplicate handlers. `session_opener` defaults to `unit_of_work.db_session` and is injectable for tests.

### `tombstone`

```
def tombstone(fn: Callable[[E], Mapping[str, Any]]) -> Mapper[E]
```

A mapper of delete events: a versioned tombstone in the owning group (§6.4 item 5); reads filter it.

### `View`

```
def View(
    name: str,
    row: type[R],
    *,
    on_stale: Literal['serve', 'source'] = 'serve',
    wait_ms: int | None = None,
) -> params.Depends
```

The dependency resolving a `ProjectionView` of projection `name`:

```
tasks: Annotated[ProjectionView[TaskRow], View("board.tasks", TaskRow)]
```

## Examples and related guides

- [Build a projection](<https://sandloom.dev/docs/modules/between/add-projection/index.md>)
- [Python API overview](<https://sandloom.dev/docs/reference/sdk/index.md>)