python-message-bus-outbox

Message bus and outbox

Event-driven architecture is a first-class option, not an exotic one. If your service publishes events to Kafka, consumes from RabbitMQ, or even just decouples in-process side effects via a local bus, this skill encodes the canonical shape.

The pattern earns its keep when:

• Multiple consumers need the same event (place order → email + analytics + webhook + warehouse)
• Side effects must survive crashes — the outbox ensures at-least-once delivery even if the publishing process dies after commit
• You're crossing process boundaries (publishing to Kafka / RabbitMQ / Redis Streams / NATS / SQS)
• You want loose coupling between use cases (saga / process manager)
• You're integrating with external services that already publish events you can subscribe to

The pattern is overkill when your service has exactly one consumer of an event AND no cross-process needs — await self._emailer.send(...) directly works fine for that case. The line is "do I need pub/sub semantics, or am I just sequencing function calls?"

This skill encodes the in-process bus + outbox + consumer/producer layout modernized for async Python.

When to use this skill

• The service is event-driven (publishes, consumes, or both)
• You're adding Kafka / RabbitMQ / Redis Streams / NATS / SQS integration
• You need at-least-once delivery via the outbox pattern
• You're building a saga / process manager
• You want to decouple in-process side effects from the use case that triggered them

Layout — where things go

Inbound and outbound are different concerns; the layout reflects that.

app/
  clients/                  # OUTBOUND: anything we call or publish TO
    openai.py
    stripe.py
    kafka.py                # KafkaProducer wrapper (publish methods only)
    rabbitmq.py             # publisher side
  consumers/                # INBOUND: anything we subscribe TO
    __init__.py             # provide_consumers() — registers consumers in lifespan
    order_consumer.py       # subscribes to "orders" topic, dispatches to services
    payment_consumer.py
  events/                   # in-process domain events (frozen dataclasses)
    order.py                # OrderPlaced, OrderCancelled
  schemas/
    events.py               # CROSS-PROCESS event wire shapes (Pydantic, for serialization)
  outbox/                   # outbox table model + dispatcher worker
    __init__.py
    model.py                # OutboxEntry (SA model)
    dispatcher.py           # background worker that drains the outbox
  handlers/                 # in-process handlers (for the local message bus)
    email_on_order_placed.py
    analytics_on_order_placed.py
  common/
    message_bus.py          # the in-process bus