Designing an event-driven system (with FastAPI)

Audience: Teams moving beyond a single API + DB into async reactions. Pairs with: RabbitMQ for commands, Kafka for domain events (this checklist).

TL;DR

1. Name bounded contexts
2. Separate commands (do work) from events (something happened)
3. Commit state → outbox → bus
4. Consumers own local models + inbox
5. Keep a living event catalog

Contents

1. Message kinds
2. Design steps
3. Topology
4. Event catalog
5. Naming and versioning
6. Delivery guarantees
7. Pseudocode
8. Process managers
9. Anti-patterns
10. Rollout plan

---

1. Message kinds

Do not treat them as interchangeable.

2. Design steps

1. Bounded contexts : who owns language and data?
2. Trigger path : API writes state, then emits command/event
3. Name events : past tense, versioned
4. Pick guarantees : outbox, inbox, keys for order
5. Catalog : producers, consumers, failure modes
6. Observe : lag, DLQ, publish rate

3. Topology

Client ──HTTP──► FastAPI (context)
 │
 ├── commands ──► RabbitMQ ──► job workers
 │
 └── outbox ──► Kafka ──► consumer A (own DB)
 └─► consumer B (own DB)

4. Event catalog

For every event:

Without a catalog, event-driven systems become tribal knowledge.

5. Naming and versioning

• Past tense for facts: OrderPlaced, not PlaceOrder
• Include version: .v1
• Payload = contract (not ORM dump)
• Stable event_id for inbox

6. Delivery guarantees

7. Pseudocode

Publish after state change

# PSEUDOCODE : Billing
async def mark_invoice_paid(invoice_id, db, outbox):
 async with db.transaction():
 invoice = await db.lock_invoice(invoice_id)
 if invoice.status == "paid":
 return invoice
 invoice.status = "paid"
 await db.save(invoice)
 await outbox.add(
 topic="billing.events",
 key=invoice.customer_id,
 payload={
 "event_id": new_uuid(),
 "type": "invoice.paid.v1",
 "invoice_id": invoice.id,
 "customer_id": invoice.customer_id,
 "amount_cents": invoice.amount_cents,
 "occurred_at": utcnow_iso(),
 },
 )
 return invoice

Consumer

# PSEUDOCODE : Notifications
async def on_invoice_paid(event, db):
 if await db.inbox_seen(event["event_id"]):
 return
 await send_receipt_email(event["customer_id"], event["invoice_id"])
 await db.inbox_insert(event["event_id"])

Combined with a job

POST export → job command (RMQ) → worker finishes
 → domain event export.completed.v1 (Kafka)
 → analytics consumer updates dashboard projection

8. Process managers

Long workflows spanning contexts:

OrderPlaced → reserve inventory command → InventoryReserved
 → charge payment command → PaymentCaptured
 → ship command

Implement as a policy service that listens to events and sends commands. Keep state in its own table; do not share DBs.

9. Anti-patterns

10. Rollout plan

1. One context, one command queue, one event type
2. Outbox for that event
3. One consumer with inbox
4. Metrics: publish rate, lag, DLQ
5. Expand catalog before the fifth event type

Related: 12 DDD · 04 Outbox · 08 FastStream