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¶

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
• Python API overview