Events and Messaging

Overview

Three messaging layers: commands (one handler), events (many handlers), and queue topics (async pub/sub). Contracts live in lexigram.contracts.events / lexigram.contracts.queue; implementations in lexigram-events (CommandBusImpl, EventBusImpl) and lexigram-queue.

Commands (CQRS)

from lexigram.events import Command, command_handler  # or implement CommandHandlerProtocol

class CreateOrder(Command):
    user_id: str
    items: list[str]

class CreateOrderHandler:
    async def handle(self, cmd: CreateOrder) -> Any:   # CommandHandlerProtocol shape
        ...

Dispatch through the bus:

from lexigram.contracts.events import CommandBusProtocol

bus = await container.resolve(CommandBusProtocol)      # CommandBusImpl bound by module
result = await bus.dispatch(CreateOrder(user_id="u1", items=[...]))

Events (Pub/Sub)

Events subclass DomainEvent from lexigram.contracts.domain — a dataclass with event_id, occurred_at, schema_version, event-sourcing fields built in:

from lexigram.contracts.domain import DomainEvent
from lexigram.events import event_handler, EventBusProtocol

class OrderCreated(DomainEvent):
    order_id: str
    user_id: str

@event_handler(OrderCreated, priority=10)
async def send_confirmation(event: OrderCreated) -> None:
    await mailer.send(event.user_id, "Order confirmed")   # resolve deps via DI, not handler args

bus = await container.resolve(EventBusProtocol)
await bus.publish(OrderCreated(order_id="o1", user_id="u1"))  # Result[None, EventError]

Handlers receive only the event. bus.subscribe(event_type, handler) registers objects implementing EventHandlerProtocol.handle(event).

Message Queues

from lexigram.contracts.queue.protocols import QueueProtocol
from lexigram.queue.consumers.consumer import MessageConsumer
from lexigram.contracts.queue.types import BusMessage

queue = await container.resolve(QueueProtocol)
await queue.publish("orders.new", BusMessage(topic="orders.new", payload={"id": "o1"}))

class OrderConsumer(MessageConsumer):
    topic = "orders.new"
    def __init__(self, queue: QueueProtocol):
        super().__init__(queue)

    async def handle(self, msg: BusMessage) -> None:
        await self.process(msg.payload["id"])

# consumer.start() subscribes; consumer.stop() unsubscribes

Transactional Outbox

The outbox lives in lexigram-sql: SQLOutboxStore.append_batch(events) writes pending events inside your DB transaction; a background OutboxPublisher(store, event_bus) polls and publishes:

from lexigram.sql.outbox import OutboxPublisher, SQLOutboxStore

store = SQLOutboxStore(provider)
await store.append_batch([OrderCreated(...)])          # same transaction as state change
publisher = OutboxPublisher(store, event_bus, poll_interval=5.0)
await publisher.start()                                 # stop() to shut down

Quick Reference

Common Mistakes

• Using events where commands fit — commands have one handler, events have many
• Writing @event_handler(OrderCreated)(fn) with extra injected params — handlers take the event only; resolve services inside or via closure
• Forgetting consumer.start() — subscribing happens there, not in the constructor
• Publishing outside a DB transaction with the outbox — events may be lost on rollback
• Not calling publisher.stop() in provider shutdown — polling task leaks