sandloom.sdk.events¶

Events: typed Event classes and on consumers, and the L2 publishing primitives.

Public API 1.0 (spec §2.2, Appendix C).

from sandloom.sdk.events 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).

Start with (beginner)¶

Advanced¶

Beginner names¶

Event¶

class Event(BaseModel)() -> None

Base class of typed events: class E(Event, channel=..., role="auto").

class NoteCreated(Event, channel="notes.note.created.v1"):
    note_id: UUID
    tenant_id: UUID

await NoteCreated(note_id=note.id, tenant_id=note.tenant_id).publish()

A class without channel is an abstract base (shared fields).

Event.publish¶

async def publish(self, *, database: str | None = None) -> None

Publish this event in the running unit (outbox, spec §5.2).

Allowed on a producer only; a view (another module's channel) is refused (SANDLOOM-C025). Raises UnitOfWorkRequiredError outside a unit.

Event.wire_schema¶

@classmethod
def wire_schema(cls) -> dict[str, Any]

The schema a producer registers and exports (the L2 form is core.register_event_schema(C, E.__name__, E.wire_schema())).

Event.meta¶

meta  (property)

Delivery metadata (EventMetaUnavailable before delivery).

Event.model_post_init¶

def model_post_init(self: BaseModel, context: Any, /) -> None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that's what pydantic-core passes when calling it.

Args:

• self: The BaseModel instance.
• context: The context.

on¶

def on(
    event: type[E],
    *,
    name: str | None = None,
    durable: bool = True,
    required: bool = True,
    dead_letter: bool = True,
    start_from: Literal['earliest', 'latest'] | datetime = 'earliest',
    group: str | None = None,
    database: str = 'default',
    tenant: str | None = 'derive',
) -> Callable[[Callable[[E], Awaitable[None]]], Callable[[E], Awaitable[None]]]

Register the decorated coroutine as a consumer of event (§5.4).

@on(NoteCreated)
async def count_note(event: NoteCreated) -> None:
    await counters.increment(event.tenant_id)

Each delivery runs in its own unit (db()), claimed once per envelope id with durable=True. tenant="derive" binds the tenant from a tenant_id: UUID field; tenant="<field>" names another UUID field; tenant=None binds none. name defaults to the function name.

Advanced names¶

EventEnvelope¶

def EventEnvelope(*args, **kwargs)

A Mapping is a generic container for associating key/value pairs.

This class provides concrete generic implementations of all methods except for getitem, iter, and len.

EventMeta¶

class EventMeta(
    id: str,
    channel: str,
    schema_version: str,
    trace_id: str | None,
    emitted_at: str,
    tenant_id: UUID | None,
) -> None

Delivery metadata of an event; mirrors the envelope, never part of the payload (event.meta on a delivered event).

PoisonEventError¶

class PoisonEventError(Exception)

Raised by a consumer handler for an event that can NEVER be applied (malformed payload): the delivery is dead-lettered at once (CRITICAL log) instead of being retried — the one way out for a dead_letter=False consumer, whose transient failures are retried forever.

publish_event¶

async def publish_event(
    channel: str,
    payload: dict[str, Any],
    schema_version: str = 'v1',
    *,
    database: str | None = None,
    required: bool | None = None,
) -> None

Publish a domain event through the transactional outbox.

Inside a UoW-managed scope (request via DbSession, db_session(), an event consumer / contract command / Celery task UoW) the event is written to the transactional outbox in the SAME transaction as the business change — it becomes visible on the bus only after commit, and is never lost or phantom-published on rollback.

Each database owns its own transactional outbox. When an endpoint holds more than one database session at once (e.g. DbSession plus Database("analytics")) pass database= to choose which database's outbox the event is written to — the business change and its event then land in the same transaction. With a single active database database may be omitted; with several active it is REQUIRED (an omitted choice raises rather than guessing and risking the wrong outbox).

Outside a UoW scope the call RAISES UnitOfWorkRequiredError (required defaults to True since 0.7). Best-effort is explicit: required=False still uses the outbox when a UoW is active, and otherwise publishes straight to the event bus — no transaction, no relay retry, lost if the broker is down. Business modules need an # sandloom: best-effort <reason> marker on such a call (sandloom check rule best-effort-effects).

publish_event_versions¶

async def publish_event_versions(
    payloads: Mapping[str, dict[str, Any]],
    *,
    database: str | None = None,
    bus: Any = None,
    required: bool | None = None,
) -> None

Publish one domain event under several versions of its channel.

payloads maps each versioned channel of ONE event to its payload, e.g. {"billing.paid.v1": old_shape, "billing.paid.v2": new_shape}. Inside a Unit of Work every version is preflighted (channel name + its own schema + size) and written to the transactional outbox in the current transaction — all versions commit together or none do. Publishing is in ascending version order. bus (scripts/tests) publishes straight to that bus. Outside a Unit of Work (and without bus) it raises like publish_event; required=False opts into the best-effort direct publish (same semantics and sandloom check marker as publish_event).

Examples and related guides¶

• Publish and consume events
• Python API overview