events-and-messaginglisted
Install: claude install-skill dbtinoy-/lexigram-framework-skills
# 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)
```python
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:
```python
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:
```python
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(EventBusPro