# `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 `. Signatures in a terminal: `sandloom docs `. Beginner names are the ones generated code and the examples use; any other name is advanced (mark its import line `# sandloom: advanced ` in generated code and examples). ## Start with (beginner) | Name | Kind | Summary | | --- | --- | --- | | [`Event`](<#event>) | class | Base class of typed events: `class E(Event, channel=..., role="auto")`. | | [`on`](<#on>) | function | Register the decorated coroutine as a consumer of `event` (§5.4). | ## Advanced | Name | Kind | Summary | | --- | --- | --- | | [`EventEnvelope`](<#eventenvelope>) | callable | A Mapping is a generic container for associating key/value pairs. | | [`EventMeta`](<#eventmeta_1>) | class | Delivery metadata of an event; mirrors the envelope, never part of the payload (`event.meta` on a delivered event). | | [`PoisonEventError`](<#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... | | [`publish_event`](<#publish_event>) | async function | Publish a domain event through the transactional outbox. | | [`publish_event_versions`](<#publish_event_versions>) | async function | Publish one domain event under several versions of its channel. | ## 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=""` 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 ` 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]()