Aggregate Pattern in Python — Core Concepts

What an aggregate is

An aggregate is a cluster of domain objects (entities and value objects) that are treated as a single unit for data changes. One entity serves as the aggregate root — the only object through which external code can interact with the cluster.

The concept comes from Eric Evans’ Domain-Driven Design. It solves a specific problem: how do you keep a group of related objects consistent when multiple parts of your system want to modify them?

The aggregate root

The aggregate root is the gatekeeper. It has three responsibilities:

1. Accept commands — External code calls methods on the root, never on internal entities
2. Enforce invariants — Before any change, the root checks business rules
3. Control access — Internal entities are not exposed for direct modification

If Order is the aggregate root and OrderLine is an internal entity, you call order.add_line(product, quantity) — not order.lines.append(new_line).

Why aggregates matter

Consistency boundaries

Without aggregates, any code can modify any object at any time. Business rules get scattered across controllers, services, and utility functions. Invariants are checked inconsistently — sometimes before saves, sometimes after, sometimes never.

Aggregates consolidate invariant checks in one place: the root. If the rule “order total cannot exceed $10,000” exists, it lives in the Order class, and it is checked every time a line is added.

Transaction boundaries

Each aggregate defines a transaction boundary. When you save an aggregate, you save the entire cluster atomically. This means the root and all its children are always in a consistent state in the database.

Cross-aggregate changes happen through eventual consistency — domain events, not a single transaction.

Concurrency boundaries

Optimistic or pessimistic locking is applied at the aggregate level. Two users can modify different orders simultaneously, but two users modifying the same order will have their changes serialized.

Aggregate design rules

Rule 1: Reference other aggregates by ID only

An Order does not hold a reference to the Customer object. It holds a customer_id. This prevents one aggregate from modifying another and keeps boundaries clean.

Rule 2: Keep aggregates small

Large aggregates cause contention (many users competing for the same lock) and performance issues (loading the entire cluster for every operation). If an aggregate has 20+ entities, consider splitting it.

Rule 3: Update one aggregate per transaction

A single use case should modify only one aggregate per database transaction. If placing an order needs to update both the Order and the Inventory, do it through events: save the Order, publish OrderPlaced, and let the inventory handler update Inventory in a separate transaction.

Python implementation sketch

from dataclasses import dataclass, field
from decimal import Decimal
from uuid import UUID

@dataclass(frozen=True)
class OrderLine:
    product_id: UUID
    name: str
    quantity: int
    unit_price: Decimal

    @property
    def total(self) -> Decimal:
        return self.unit_price * self.quantity

@dataclass
class Order:  # Aggregate root
    id: UUID
    customer_id: UUID
    _lines: list[OrderLine] = field(default_factory=list)
    status: str = "draft"

    MAX_TOTAL = Decimal("10000")

    @property
    def lines(self) -> tuple[OrderLine, ...]:
        return tuple(self._lines)  # Read-only view

    @property
    def total(self) -> Decimal:
        return sum(line.total for line in self._lines)

    def add_line(self, product_id: UUID, name: str,
                 quantity: int, unit_price: Decimal) -> None:
        if self.status != "draft":
            raise ValueError("Cannot modify a non-draft order")
        new_line = OrderLine(product_id, name, quantity, unit_price)
        projected_total = self.total + new_line.total
        if projected_total > self.MAX_TOTAL:
            raise ValueError(f"Order total would exceed {self.MAX_TOTAL}")
        self._lines.append(new_line)

    def remove_line(self, product_id: UUID) -> None:
        if self.status != "draft":
            raise ValueError("Cannot modify a non-draft order")
        self._lines = [l for l in self._lines if l.product_id != product_id]

    def submit(self) -> None:
        if not self._lines:
            raise ValueError("Cannot submit an empty order")
        self.status = "submitted"

Key design points:

• _lines is private — external code uses add_line() and remove_line()
• The lines property returns a tuple (immutable view)
• Every mutation checks invariants before applying changes

Common misconception

“Every entity should be its own aggregate.” This leads to aggregates that are too small, requiring complex cross-aggregate coordination. Group entities that share invariants into the same aggregate. OrderLine only makes sense in the context of an Order, so they belong together.

Identifying aggregate boundaries

Ask: “What objects must be consistent together, immediately?” Those objects belong in one aggregate. Objects that can be eventually consistent belong in separate aggregates.

If you can tolerate “customer address updated 2 seconds after order was placed,” then Customer and Order are separate aggregates. If “order line total must always match order total” is a hard rule, then OrderLine belongs inside the Order aggregate.

The one thing to remember: Aggregates group objects that must stay consistent together and funnel all changes through a root that enforces business rules — keeping your domain logic centralized and reliable.