# Add a consumer **Goal:** your module reacts to an event another module publishes, such as counting items when `tasks.item.created.v1` arrives. **Use it** for work that may happen after the producer committed: counters, read models, notifications, starting your own process. **Do not use it** when the producer needs your answer before it commits (that is a [contract]()), or to keep a list page in sync from several channels with versions (that is a [projection](), which generates its consumers for you). Delivery is asynchronous and at-least-once; there is no ordering guarantee between events. ## Prerequisites - Your module exists ([Add a module]()). - The producer's schema is known to the generator, in one of two places: - the producer is a module in the same app (its `Event` class is found in the tree); - or the producer's exported schema file is in your shared DTO package, `/shared/events/.json` (written by `sandloom generate` in the producer's repository). Otherwise you pass the fields yourself with `--field`, and the view is marked unverified. - If `consumers.py` already exists, it must parse and import `Event, on` from `sandloom.sdk` (and the types the fields need). ## Steps ### 1\. Run the generator in the consuming module ``` sandloom add consumer board notes.item.created.v1 ``` When no producer schema is found, the generator refuses and asks for the fields: ``` sandloom add consumer board billing.invoice.paid.v1 \ --field invoice_id:UUID --field tenant_id:UUID --field amount:int ``` The generator **appends** a view class and an `@on` handler to `/consumers.py` (created when absent). ### 2\. Read what it wrote Both runs above, in one file: ``` """Consumers of the board module.""" from uuid import UUID from sandloom.sdk import Event, on class ItemCreated(Event, channel="notes.item.created.v1"): """A view: fields of the producer (notes/events.py).""" item_id: UUID tenant_id: UUID @on(ItemCreated) async def on_item_created(event: ItemCreated) -> None: """Runs once per event (durable), with the event's tenant bound.""" class InvoicePaid(Event, channel="billing.invoice.paid.v1"): # unverified: no producer schema was found """A view: fields from --field.""" invoice_id: UUID tenant_id: UUID amount: int @on(InvoicePaid) async def on_invoice_paid(event: InvoicePaid) -> None: """Runs once per event (durable), with the event's tenant bound.""" ``` - **A view, not a producer.** The channel's first segment (`notes`) is another module, so the class is a *view* of that channel: it is not registered as a schema, `.publish()` on it is refused, and payload fields it does not declare are ignored. You may delete fields you do not need; never add one the producer does not send. - **Verified or not.** A view built from the producer's schema is checked against it. A view built from `--field` carries the comment `# unverified: no producer schema was found` and is reported as unverified until the producer's schema file is available. - **The handler.** `@on(E)` registers the function as a durable consumer named after the function. Each delivery runs in its own unit of work; `db()` is that unit's session. The envelope id is claimed in the same unit, so a redelivered event applies its effect once. The tenant is bound from the event's `tenant_id` field. ### 3\. Write the effect Fill the handler with the effect, through your module's services. The tutorial app's `stats` module counts items: ``` class ItemCreated(Event, channel="tasks.item.created.v1"): """A view: fields of the producer (tasks/events.py).""" item_id: UUID tenant_id: UUID @on(ItemCreated) async def on_item_created(event: ItemCreated) -> None: """Runs once per event (durable), with the event's tenant bound.""" await services.record_item() ``` ``` async def _counter() -> StatsItemCount | None: # TenantScoped rows are filtered by the bound tenant: at most one row. result = await db().execute(select(StatsItemCount)) return result.scalar_one_or_none() async def record_item() -> None: counter = await _counter() if counter is None: db().add(StatsItemCount(items=1)) else: counter.items += 1 await db().flush() async def item_count() -> int: counter = await _counter() return 0 if counter is None else counter.items ``` Keep the effect inside the handler's unit of work. An effect outside the database (an email, a call to a provider) is not covered by the claim: make it idempotent on the event's id or the business key. To stop retrying an event that can never succeed, raise `PoisonEventError` (from `sandloom.sdk.events`): the event goes to the dead-letter queue at once. Any other exception rolls back the unit and the event is redelivered, then dead-lettered after the delivery budget. ### 4\. Options you may need `@on` takes the options of the underlying consumer registration. The defaults are right for most handlers: | Option | Default | When to change it | | --- | --- | --- | | `name` | the function name | two handlers of one channel in one module need distinct names | | `durable` | `True` | `False` only for an effect that is harmless when repeated | | `required` | `True` | `False` if the app may start without this consumer running | | `dead_letter` | `True` | `False` to retry forever with capped backoff (a `PoisonEventError` still dead-letters) | | `start_from` | `"earliest"` | `"latest"` or a datetime for a new consumer that must skip history | | `group` | `None` (`:`) | pin an explicit group id, for example to drain a group created under another name | | `database` | `"default"` | the module's tables live in another database | | `tenant` | `"derive"` | `""` for another UUID field, `None` to bind no tenant | ## Verify Deliver the event in a test with `deliver`. With `times=2` the same envelope id arrives twice, which proves the effect is applied once: ``` async def test_a_redelivered_event_is_counted_once(client: Any, db: Any) -> None: tenant = uuid4() _, headers = as_user(tenant_id=tenant) event = ItemCreated(item_id=uuid4(), tenant_id=tenant) await deliver(event, times=2, session=db) # the same envelope id twice response = await client.get("/api/v1/stats/count", headers=headers) assert response.status_code == 200, response.text assert response.json() == {"items": 1} async def test_counts_are_per_tenant(client: Any, db: Any) -> None: await deliver(ItemCreated(item_id=uuid4(), tenant_id=uuid4()), session=db) _, headers = as_user(tenant_id=uuid4()) # another tenant response = await client.get("/api/v1/stats/count", headers=headers) assert response.json() == {"items": 0} ``` ``` sandloom test stats ``` ## Errors you may hit - [SANDLOOM-B016](): the handler's parameter is not the event class, a consumer name is used twice, or the tenant field does not exist on the event. - [SANDLOOM-B015](): the view's channel is not versioned, or a field is not a wire type. - [SANDLOOM-C025](): the view does not match the producer's schema, is unverified, or calls `.publish()`. - [SANDLOOM-R001](): the handler touched a tenant table with no tenant bound (the event has no `tenant_id`, or `tenant=None`). ## Related - [Events and consumers](): delivery, the dead-letter queue, consumer groups. - [Add an event](): the producer side. - [Add a projection](): a read model fed by events. - [Testing modules]().