Using DDD with FastAPI (and workers)

Audience: Teams whose FastAPI apps outgrew “fat routes + shared models.” DDD is: language, boundaries, and ownership : not a folder religion and not “event sourcing required.”

TL;DR

Contents

1. Why DDD helps
2. Core ideas
3. Suggested layout
4. Aggregates
5. Application use cases
6. Mapping to jobs and events
7. Contexts vs packages
8. Tactical patterns
9. Testing
10. Anti-patterns
11. Adoption path

---

1. Why DDD helps

Without boundaries, FastAPI codebases often grow:

• Fat routers with business rules
• One models.py for the company
• Jobs that UPDATE five “domains” in one transaction

DDD pushes invariants next to the data, explicit integration, and shared language.

2. Core ideas

3. Suggested layout

app/
 contexts/
 billing/
 domain/ # entities, value objects, domain events, errors
 application/ # use cases
 adapters/
 api/ # FastAPI routers + Pydantic
 persistence/ # ORM, repositories
 messaging/ # publish commands/events
 notifications/
 domain/
 application/
 adapters/
 workers/ # process entrypoints → application use cases
 main.py

Start as a modular monolith. Split deployables when scale or team ownership demands it.

4. Aggregates

# PSEUDOCODE : domain (no FastAPI imports)
@dataclass
class Invoice:
 id: str
 status: str # draft | issued | paid | void
 amount_cents: int

 def pay(self, at: datetime) -> list[DomainEvent]:
 if self.status == "paid":
 return []
 if self.status != "issued":
 raise DomainError("only issued invoices can be paid")
 self.status = "paid"
 return [InvoicePaid(self.id, self.amount_cents, at)]

Rules live on the aggregate, not only in route validators.

5. Application use cases

# PSEUDOCODE
async def pay_invoice(cmd: PayInvoice, repo: InvoiceRepo, outbox: Outbox) -> None:
 async with unit_of_work():
 invoice = await repo.get(cmd.invoice_id)
 events = invoice.pay(utcnow())
 await repo.save(invoice)
 for e in events:
 await outbox.add(e.to_integration_event())

Thin route:

@router.post("/invoices/{invoice_id}/pay", status_code=204)
async def pay(invoice_id: str, svc=Depends(get_billing_app)):
 await svc.pay_invoice(PayInvoice(invoice_id=invoice_id))

Workers call the same use case (or a command variant) with a messaging adapter.

6. Mapping to jobs and events

POST /invoices/{id}/pay
 → Billing application service
 → Invoice.pay()
 → outbox: invoice.paid.v1
 → Kafka
 → Notifications policy → command SendReceiptEmail → RabbitMQ
 → email worker

7. Contexts vs packages

• Do not import billing.domain from notifications.domain
• Notifications may depend on integration event schemas, not Billing ORM
• Shared kernel only for true shared primitives (Money, TenantId) if agreed

8. Tactical patterns

You can practice DDD without event sourcing.

9. Testing

1. Domain tests : pure aggregate rules
2. Application tests : fake repo + fake outbox
3. Adapter tests : HTTP → command; message → command
4. Contract tests : event payload schema

10. Anti-patterns

11. Adoption path

1. Name 2-3 contexts on one page
2. One aggregate with real invariants
3. Thin the route; rules on the aggregate
4. One domain event via outbox
5. One consumer + inbox in another context
6. Only then split deployables

Related: 11 Event-driven · 04 Outbox