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, <app>/shared/events/<channel>.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 <module>/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:

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.