What DDD actually is
Domain-Driven Design is Eric Evans' answer to a specific problem: once a business's rules get complicated enough, the code that expresses those rules deserves to be a first-class citizen of the codebase, modeled in its own vocabulary, rather than dissolved into database columns and controller logic. The tactical piece of DDD — the piece this whole article is about — is a small vocabulary of building blocks: entities (objects with identity and a lifecycle, like an Order), value objects (immutable, identity-less, like an OrderItem's quantity and price), and aggregates (a cluster of entities and value objects with one root that's the only door in — nothing outside touches an OrderItem except through its parent Order). Layering — domain, application, infrastructure, presentation — is the architectural consequence of taking that vocabulary seriously: if the domain model is supposed to be the authority on business rules, it can't also depend on a database driver or a web framework, or that authority quietly moves to whatever ORM query happens to run first.
It's worth being upfront that DDD isn't the only way to structure this, and for a lot of systems it's the wrong one. Martin Fowler's own framing is that a Transaction Script — one procedure per use case, talking to the database directly or through a thin wrapper, no separate domain model at all — is perfectly fine for the majority of business systems where the logic per operation really is a short, linear checklist. Microsoft's own architecture guidance says the same thing from the other direction: reach for DDD when a domain is genuinely complex, not as a default. An Active Record or "Smart UI" style, where a model class is both the business object and its own persistence — Django models and Rails' ActiveRecord are the canonical examples — trades exactly the same thing away: less ceremony per change, at the cost of business rules and database concerns living in the same class, which is fine until the rules outgrow a single table's shape.
A third alternative sits further out: event sourcing, where a domain's state isn't stored directly at all — instead, every state change is appended as an immutable event, and the current state is whatever you get from replaying the event log from the start. Paired with CQRS (writes go through the event-sourced model, reads come from a separately maintained projection) and a message queue or pub/sub broker to fan those events out to other services, this is the shape a lot of event-driven microservice systems take. It's the natural fit for anything that's already conceptually a ledger — a bank account balance, or a stock-on-hand figure — where the "current state" was never really a single number to begin with, just the running total of every deposit, withdrawal, receipt, and sale that ever happened to it. A bank doesn't overwrite your balance column on every transaction; it appends a movement and the balance is sum(movements), which is exactly what makes "what was my balance last Tuesday" a normal question to ask instead of an impossible one. Current stock in this ERP already works the same way conceptually — a product's stock level is the net of every order received, every sale, every manual adjustment — even though today it's persisted as a running column updated in place rather than derived by replaying a movement log.
I haven't used full event sourcing here — the orders module below stores current state directly and dispatches events only as a side effect, not as the source of truth — but it's worth sketching what it would look like, because the domain events this module already has (OrderCreated, OrderReceived, OrderCancelled) are one design decision away from becoming exactly that:
# Not what this module does — event sourcing, for contrast.
# The event *is* the state; there's no `orders` row to update.
class OrderReceived:
order_id: int
received_at: datetime
def apply(order: OrderState, event: OrderReceived) -> OrderState:
return replace(order, status=OrderStatus.RECEIVED,
delivery_date=event.received_at)
# Rebuilding "the current order" means replaying its whole history:
def rebuild(events: list[DomainEvent]) -> OrderState:
state = OrderState.empty()
for event in events:
state = apply(state, event)
return state
# In a pub/sub system, that same OrderReceived event also gets
# published to a broker (Kafka, SNS, RabbitMQ) so that, say, a
# separate billing service can react without this module knowing
# billing exists at all.
broker.publish("orders.received", OrderReceivedIntegrationEvent(order_id=order.id))That model buys a full audit trail for free (every historical state is reconstructible) and lets independent services subscribe to what happened without coupling to how it's stored — genuinely valuable for a payment ledger or a multi-service pipeline. It also costs real complexity: replaying thousands of events to answer "what does this order look like right now" needs snapshotting to stay fast, and querying "all orders above $X" isn't a WHERE clause anymore, it's a projection you have to build and keep in sync. For a single-service module where the honest current-state question ("is this order still pending?") is asked far more often than "what was its full history," that trade wasn't worth making — which is exactly the kind of judgment call DDD's own literature says to make deliberately rather than by default.
The layered split without event sourcing still earns its ceremony, and the clearest single reason is unit testing. Because Order and procurement_policy.py below are plain Python objects with no framework or database import, a test for "can a received order be cancelled?" is a synchronous function call against a dataclass — no test database, no fixtures, no async test client. A change to a business rule shows up as a failing assertion in milliseconds, not as an integration test that has to provision a schema first. That property degrades in every alternative above roughly in proportion to how much the model is allowed to know about persistence: Active Record couples the two directly, Transaction Script never separates them to begin with, and only a domain layer with zero I/O keeps the rule and the round-trip to a database from ever being the same test.
Let's use a real module as the example throughout rather than staying abstract: the purchase-orders module in this ERP's inventory domain, at modules/inventory/orders/. Here's its full folder structure, unedited — the four DDD layers are literally the four top-level folders, and every code sample below is pulled from one of these files:
orders/
├── domain/
│ ├── order.py # Order aggregate root
│ ├── order_item.py # OrderItem value object
│ ├── enums.py # OrderStatus, ProcurementDecision
│ ├── events.py # OrderCreated, OrderReceived, OrderCancelled
│ ├── exceptions.py # OrderNotFoundError, InvalidOrderStatusError, ...
│ ├── procurement_policy.py # pure functions: triggers, blockers, decision
│ └── purchase_recommendation.py
├── application/
│ ├── service.py # OrderService — UoW orchestration
│ ├── purchase_recommendation_service.py
│ ├── bootstrap.py # composition root / factory functions
│ └── notifications.py
├── infrastructure/
│ ├── repository.py # OrderRepository (SQLAlchemy)
│ ├── models.py # OrderORM, OrderItemORM, OrderHistoryORM
│ ├── mapper.py # Order <-> ORM translation
│ ├── query_service.py # read-only CQRS-lite path
│ ├── notification.py / email_builder.py
└── presentation/
├── router.py # FastAPI endpoints
├── schemas.py # Pydantic request/response models
└── dependencies.py # FastAPI DI providersThe problem
"Layered architecture" is easy to describe and easy to fake. A folder named domain/ that quietly imports SQLAlchemy, a "service" that's really a thin wrapper around a repository call with no orchestration logic of its own, a router that reaches past its own application layer straight into a repository because it was faster that Friday afternoon — all of it can still be labelled domain-driven design in a slide deck. The only way to know whether a codebase actually respects layering is to open the folders and read what's inside them, layer by layer, and check that the dependency arrows actually point where the diagram says they do.
The purchase-orders module in this ERP's inventory domain is a good test case precisely because it isn't a toy CRUD example. It has state transitions with real business consequences (an order can't be un-received), a procurement-recommendation engine that combines stock signals with forecast data, and domain events that trigger notifications — enough moving parts that a lazy layering would have shown cracks. It didn't have to be forced into a four-layer template for this article; the four layers are literally the four top-level folders under orders/.
The alternatives
- One flat module — models, routes, and logic in a couple of files. Fastest to write, and fine for a handful of endpoints. It breaks down exactly where this module lives: status-transition rules, procurement policy, notification side effects, and CQRS-style read paths all end up interleaved in the same files, and there's no boundary stopping a router from mutating an ORM row directly and skipping the business rule that was supposed to guard it.
- Service layer only, no explicit domain layer. A common middle ground — routers call "services" that talk to the ORM directly, with business rules expressed as `if` statements scattered across service methods. It's better than nothing, but the rules aren't unit-testable without a database, and there's no single object that owns "what does a valid order look like," so the same invariant (e.g. "only PENDING orders can be edited") tends to get re-checked, or forgotten, in more than one place.
- Full four-layer DDD split, with dependency arrows enforced by folder structure. Domain holds pure business rules with zero framework imports; application orchestrates transactions and calls the domain; infrastructure implements persistence against the domain's own types; presentation exposes HTTP and depends on application only. More ceremony per change — a status transition touches an aggregate method, a service method, and a mapper — but every rule has exactly one home, and that rule is testable without a running database. This is what the orders module actually does.
The decision
Walking the four folders from the tree above, layer by layer — starting with domain, since it's the one every other layer depends on and none of them may depend back on.
Domain: an aggregate that enforces its own rules
The Order aggregate is a plain @dataclass — no Base, no SQLAlchemy, no FastAPI. Status transitions are gated by an explicit adjacency map rather than scattered if checks, and the module's status values are stored as the actual Portuguese labels the business uses:
class OrderStatus(StrEnum):
PENDING = "Pendente"
RECEIVED = "Entregue"
CANCELLED = "Cancelado"
@dataclass
class Order:
VALID_TRANSITIONS: ClassVar[dict[OrderStatus, set[OrderStatus]]] = {
OrderStatus.PENDING: {OrderStatus.RECEIVED, OrderStatus.CANCELLED},
OrderStatus.RECEIVED: set(),
OrderStatus.CANCELLED: set(),
}
def can_transition_to(self, new_status: OrderStatus) -> bool:
return new_status in self.VALID_TRANSITIONS.get(self.status, set())
def mark_received(self) -> None:
assert self.id is not None, "mark_received requires a persisted order"
self._change_status(OrderStatus.RECEIVED)
self.delivery_date = self._clock.today()
self._events.append(OrderReceived(order_id=self.id, ...))Two details are worth calling out because they're easy to get wrong. First, OrderItem is a frozen value object — quantity and unit price validate themselves in __post_init__, and the aggregate mutates an order's items by replacing the whole tuple rather than editing an item in place. Second, time isn't read from datetime.now() inside the aggregate; it comes through an injected Clock protocol, which is what lets mark_received be unit-tested with a fixed date instead of a real clock:
class Clock(Protocol):
def today(self) -> date: ...
def now(self) -> datetime: ...
@dataclass
class SystemClock:
def today(self) -> date:
return datetime.now(get_app_timezone()).date()The module's most interesting domain code isn't even about orders directly — it's procurement_policy.py, a set of pure functions with no class at all, deciding whether a product needs a purchase recommendation. It deliberately separates two concerns that change at different rates: triggers ("does this product need buying," tuned by operations) from blockers ("are we forbidden from buying right now," a correctness constraint):
def evaluate_procurement_state(
stock_status: StockStatus,
risk_level: RiskLevel,
has_pending_order: bool,
last_stock_update: datetime | None,
last_confirmed_delivery: datetime | None,
) -> ProcurementDecision:
if not is_procurement_triggered(stock_status, risk_level):
return ProcurementDecision.NOT_NEEDED
signal = has_new_consumption_signal(last_stock_update, last_confirmed_delivery)
if not is_procurement_allowed(has_pending_order, signal):
return ProcurementDecision.BLOCKED
return ProcurementDecision.RECOMMENDReturning a four-state enum (NOT_NEEDED, WATCH, RECOMMEND, BLOCKED) instead of a boolean is itself a small design decision: a caller can tell "no purchase needed" apart from "purchase needed but currently blocked," which a plain bool would have collapsed into the same value.
A domain layer that can't be unit-tested without a database isn't a domain layer — it's a service layer wearing the folder name.
Application: orchestration, transactions, no business rules of its own
OrderService owns the transaction boundary and event dispatch — nothing else. Its own docstring says as much: "business rules live on the Order aggregate, not here." The transaction boundary itself is a plain async context manager, one per bounded context, that commits on a clean exit and rolls back on any exception:
class InventoryUnitOfWork:
def __init__(self, session: AsyncSession) -> None:
self._session = session
self.orders = OrderRepository(session)
self.products = ProductRepository(session)
self.suppliers = SupplierRepository(session)
async def __aenter__(self) -> Self:
await self._session.begin()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
if exc_type is not None:
await self._session.rollback()
else:
await self._session.commit()That's what makes async with self._uow: meaningful rather than decorative — every repository reachable off the UoW shares the same session, so a write to self._uow.orders and a read from self._uow.suppliers inside the same block are part of one atomic unit, and any exception raised anywhere in the block rolls all of it back. Creating an order shows the pattern: open the unit of work, allocate a sequence-based order number inside that same transaction (so a rollback releases the sequence), build the domain aggregate through its factory method, save it, then drain and dispatch the events only after the transaction has closed:
async def create_order(self, supplier_id, items, expected_delivery_date):
async with self._uow:
order_number = await self._build_order_number()
supplier = await self._uow.suppliers.get_by_id(supplier_id)
domain_items = [OrderItem(product_id=i.product_id,
quantity=Decimal(str(i.quantity)),
unit_price=Decimal(str(i.unit_price)))
for i in items]
entity = Order.create_order(supplier_id=supplier_id,
order_number=order_number,
items=domain_items,
expected_delivery_date=expected_delivery_date)
events = entity.pull_events()
saved = await self._uow.orders.save(entity)
self._background_tasks.add_task(self._dispatcher.dispatch_all, events)
return self._build_response(saved, supplier_name)The events are pulled off the aggregate before commit but dispatched after — inside the async with self._uow block the transaction is still open; once that block exits cleanly the commit has happened, and only then does background_tasks.add_task schedule the dispatcher. If the transaction rolls back, the events never fire, because they were never handed to the dispatcher in the first place. Nothing about OrderService above ever constructs its own dependencies — a separate file owns that:
# application/bootstrap.py — composition root
def build_order_service(
db: AsyncSession, background_tasks: BackgroundTasks
) -> OrderService:
uow = UnitOfWorkFactory(db).inventory()
query = OrderQueryService(db)
dispatcher = _build_dispatcher(query)
return OrderService(uow, query, dispatcher, background_tasks)
def _build_dispatcher(query: OrderQueryService) -> EventDispatcher:
dispatcher = EventDispatcher()
handler = OrderNotificationHandler(query)
dispatcher.register(OrderCreated, handler.on_order_created)
dispatcher.register(OrderReceived, handler.on_order_received)
dispatcher.register(OrderCancelled, handler.on_order_cancelled)
return dispatcherbootstrap.py's own module docstring calls it out directly: "usable from HTTP, agents, workers, CLI." build_order_service takes a raw AsyncSession and a BackgroundTasks instance as its only inputs — nothing FastAPI-specific about the wiring itself, just the two objects a caller has to supply. A request handler gets both from FastAPI's dependency injection; a worker script would construct them directly. The same factory function is what makes an OrderService callable from the HTTP layer today and equally callable from a worker or a script tomorrow, since nothing about the wiring depends on a request being in flight.
Where side effects run: inside the transaction, after it, or on a queue
An order being marked received triggers an unrelated side effect — a notification email to the supplier. Where that side effect runs relative to the database transaction is its own small design decision, and the three options aren't equivalent:
- Inside the transaction. Send the email (or call the notification service) from directly inside the
async with self._uowblock, before it commits. This buys the strongest consistency guarantee — the email only "happens" if the write actually commits — but it's the wrong trade for this case. It holds the transaction open for as long as the SMTP call takes, adding that latency directly to the request's response time, and it ties something that shouldn't matter to correctness — whether an email server answered in time — to whether an order can be marked received at all. A supplier's mail server timing out is not a reason to fail a status transition that already happened in the business's terms. - After the transaction, off the request path. Pull the events off the aggregate, let the transaction commit, then hand them to a dispatcher that runs once the response has already been sent. This is what the code actually does — no added latency, and consistency is preserved because a rolled-back transaction never hands its events over in the first place. The cost: it isn't durable. A crash or a failing handler between commit and dispatch loses the notification silently.
- On a queue, consumed by a separate worker. Publish the event to a broker (Redis/RQ, SQS, RabbitMQ) instead of scheduling an in-process callback. This is the durable version of option two: the event survives a process restart because it lives in the broker, not in a Python list, and a failed delivery gets retried with backoff instead of just logged. It costs real operational surface — a broker to run, a worker to deploy, idempotency to design for since retries can run a handler more than once.
What the code actually does is the middle option — dispatch after the transaction closes, off the request path:
class EventDispatcher:
def __init__(self) -> None:
self._handlers: dict[type, list[Callable]] = defaultdict(list)
def register(self, event_type: type, handler: Callable) -> None:
self._handlers[event_type].append(handler)
async def dispatch(self, event: Any) -> None:
for handler in self._handlers[type(event)]:
try:
await handler(event)
except Exception:
logger.exception("Handler %s failed for event %s",
handler.__qualname__, type(event).__name__)
async def dispatch_all(self, events: Iterable[Any]) -> None:
for event in events:
await self.dispatch(event)background_tasks.add_task is FastAPI's own mechanism: the callable runs after the response has already been sent, in the same process, on the same event loop. That solves the latency problem — a customer clicking "receive" isn't waiting on an SMTP round-trip — and the commit-before-dispatch ordering solves the consistency problem, since a rolled-back transaction never hands its events to the dispatcher in the first place. It does not solve durability. dispatch's try/except Exception is doing real work here: without it, one failing handler would crash a background task with nothing watching it. With it, the failure is logged — and then the event is gone. If the process restarts between commit and dispatch, or the notification handler raises because the supplier's mail server is down, that email is never sent and nothing retries it, because there was never anything durable holding the intent to send it — only an in-memory Python list, for the lifetime of one request.
The more correct shape for this specific problem — one unrelated to whether the domain/application split itself is right — is a real message queue: publish the event to Redis/RQ, SQS, or RabbitMQ instead of scheduling an in-process callback, and let a separate worker process consume it:
# Not what this module does — illustrative, for contrast with dispatch_all above.
async def dispatch_all(self, events: Iterable[Any]) -> None:
for event in events:
await self._queue.enqueue(
"orders.notify_supplier",
payload=event.to_dict(),
retry=Retry(max=5, interval=[10, 30, 60, 300, 900]),
)
# A separate worker process, not the request/response cycle, does this:
@worker.task("orders.notify_supplier", max_retries=5)
async def notify_supplier(payload: dict) -> None:
await send_supplier_email(payload)A queue turns "logged and lost" into "retried with backoff, and parked in a dead-letter queue for inspection if it keeps failing" — the event survives a process restart because it's durable in the broker, not in a Python list. It also adds real operational surface: a broker to run and monitor, a worker process to deploy and scale independently of the API, and idempotency to think about, since retries mean a handler might run more than once for the same event. For a solo-built system at this traffic level, BackgroundTasks plus a logged exception was the trade actually made — good enough that a failed notification is a rare, loggable event rather than a silent one, but explicitly not the durable version of this pattern.
Infrastructure: repository, ORM models, and the mapper between them
models.py is where the table shape actually lives — SQLAlchemy columns, foreign keys, and a status CheckConstraint that mirrors the domain's own OrderStatus enum so an invalid status can't even be written directly against the database, bypassing the aggregate entirely:
class OrderORM(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
order_number: Mapped[str] = mapped_column(String(50), unique=True)
supplier_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("suppliers.id"))
status: Mapped[str] = mapped_column(String(20))
order_date: Mapped[date] = mapped_column(Date)
total_value: Mapped[Decimal] = mapped_column(Numeric(12, 2))
items: Mapped[list[OrderItemORM]] = relationship(
"OrderItemORM", back_populates="order",
cascade="all, delete-orphan", lazy="raise",
)
__table_args__ = (
CheckConstraint(status.in_([s.value for s in OrderStatus]),
name="check_order_status"),
)This class is never imported outside infrastructure/ — no service, no router, no test of a business rule ever sees OrderORM directly. The domain's Order and the persisted OrderORM are two different classes on purpose — the mapper is the only place that knows how to translate between them, and it lives in infrastructure/ specifically because it's the file that's allowed to import both:
class OrderMapper:
@staticmethod
def to_domain(orm: OrderORM) -> Order:
return Order(
id=orm.id, order_number=orm.order_number,
status=OrderStatus(orm.status), order_date=orm.order_date,
supplier_id=orm.supplier_id,
_items=[OrderItem(product_id=i.product_id,
quantity=Decimal(str(i.quantity)),
unit_price=Decimal(str(i.unit_price)))
for i in orm.items],
)OrderRepository.save makes the insert/update split explicit rather than relying on the ORM's session to guess intent: a new order (id is None) goes through OrderMapper.to_orm and an insert; an existing one is fetched fresh, mutated field-by-field from the domain entity, and its item collection replaced wholesale. The ORM relationships all declare lazy="raise" — an accidental lazy-load outside an explicit selectinload throws instead of silently issuing an extra query, which is what keeps N+1s from hiding in code that looks fine in a code review.
Reads take a separate path entirely. OrderQueryService (also in infrastructure/) selects exactly the columns a list screen or an aggregate chart needs and returns response schemas directly, never full domain aggregates — a CQRS-lite split that keeps the read side from paying the cost of loading and reconstructing a full Order just to render a table row.
Presentation: a router with no business logic in it
The router's job is HTTP translation only — parse query params, call a service method, return what it gets back:
@router.patch("/{order_id}/receive", response_model=OrderResponse)
async def receive_order(
order_id: int, service: OrderService = Depends(get_order_service)
):
return await service.receive_order(order_id)
@router.patch("/{order_id}/cancel", response_model=OrderResponse,
dependencies=[Depends(require_admin)])
async def cancel_order(
order_id: int, service: OrderService = Depends(get_order_service)
):
return await service.cancel_order(order_id)There's no if order.status != "Pendente" anywhere in this file — that guard lives on the aggregate, where it belongs, and the router never sees it. schemas.py is where the Pydantic request/response contracts live, separate from both the domain's OrderItem value object and the ORM's OrderItemORM; OrderItemCreate, OrderItemUpdate, and OrderItemResponse are three distinct shapes for three distinct directions of data flow, even though they share most of their fields, because a create payload, an update payload, and a response don't have the same optionality rules. dependencies.py is the thin FastAPI-specific glue — it just calls into bootstrap.py's factory functions with a request-scoped database session, keeping the wiring itself outside the router.
What doesn't fit the clean template
Being honest about the actual folders matters more than making the diagram pretty. Two things don't fit a textbook split. First, the domain layer's exception classes carry an http_status class attribute (OrderNotFoundError.http_status = 404) — technically an HTTP concept leaking into domain code, treated here as transport-agnostic classification metadata rather than a framework dependency, and explained at length in a companion piece on how those exceptions get translated to HTTP responses. Second, the application layer directly imports presentation-layer schemas (OrderItemCreate, OrderUpdate) as its own method parameter and return types, rather than defining separate application-layer DTOs — a pragmatic shortcut that saves a translation step at the cost of a small crack in the "arrows only point inward" rule. Neither breaks the module in practice, but neither survives a strict reading of the dependency rule either.
The outcome
The Order aggregate and procurement_policy.py can be exercised in unit tests with plain Python objects and a fake Clock — no database, no FastAPI test client, no fixtures beyond constructing dataclasses directly. Status-transition bugs (can a received order be cancelled? can an item be added after the fact?) show up as a failing assertion against VALID_TRANSITIONS or assert_editable, not as an integration test that has to spin up a schema first. The four folders are also what make the module's public surface legible from the outside: anything that needs to call into orders — a router, a background job, another module's service — goes through application/bootstrap.py's factory functions and never touches OrderORM or OrderRepository directly, which is the actual, checkable version of "modules interact through their application layer" rather than a claim that only holds until someone's in a hurry.