events-and-messaginglisted
Install: claude install-skill dbtinoy-/lexigram-framework-skills
# Events and Messaging
## Overview
Lexigram provides three messaging primitives: commands (one handler), events (many handlers), and queues (async pub/sub). Built on `lexigram-events` and `lexigram-queue`.
## Commands (CQRS)
```python
from lexigram.events import Command, CommandHandlerProtocol
class CreateOrder(Command):
user_id: str
items: list[Item]
class CreateOrderHandler(CommandHandlerProtocol[CreateOrder, Result[Order, DomainError]]):
def __init__(self, repo: OrderRepositoryProtocol):
self.repo = repo
async def handle(self, cmd: CreateOrder) -> Result[Order, DomainError]:
order = Order(user_id=cmd.user_id, items=cmd.items)
return await self.repo.save(order)
```
## Events (Pub/Sub)
```python
from lexigram.events import DomainEvent, event_handler
class OrderCreated(DomainEvent):
order_id: str
user_id: str
@event_handler(OrderCreated)
async def send_confirmation(event: OrderCreated, mailer: MailerProtocol) -> None:
await mailer.send(event.user_id, "Order confirmed")
```
## Buses
```python
from lexigram.events import CommandBus, EventBus
bus = await container.resolve(CommandBus)
result = await bus.dispatch(CreateOrder(user_id="u1", items=[...]))
bus = await container.resolve(EventBus)
result = await bus.publish(OrderCreated(order_id="o1", user_id="u1"))
```
## Message Queues
```python
from lexigram.queue import QueueProtocol, BusMessage
queue = await container.resolve(QueueProtocol)
await queue.publish("o