Aggregates: Consistency Boundaries

The Problem with Tangled Objects

In a complex system, objects are connected. An Order has a list of OrderItems. Each OrderItem references a Product. The Order has a ShippingAddress and a PaymentMethod.

If multiple parts of your system try to modify these related objects simultaneously without coordination, you can corrupt your data. One thread adds an item to an Order while another calculates the Order total — the total ends up wrong. One service marks an item as shipped while another cancels the order — the state becomes inconsistent.

You need a way to define: "these objects belong together and must always be changed together, atomically."

The Aggregate Pattern

An Aggregate is a cluster of domain objects (Entities and Value Objects) that are treated as a single unit for the purpose of data changes. Every aggregate has one designated Entity called the Aggregate Root.

The rules are strict:

1. Single Root: The Aggregate Root is the only object in the cluster that outside code is allowed to hold a reference to. You cannot reach inside an aggregate and grab an OrderItem directly.
2. All Access Through the Root: If you want to add an item to an order, you call order.add_item(...). The root is the gatekeeper. It enforces all business rules (called invariants) before allowing the change.
3. Atomic Persistence: The entire aggregate is saved to the database in a single transaction. You never save half an aggregate.
4. One Aggregate Per Transaction: As a rule of thumb, a single business operation should only modify one aggregate. If you find yourself modifying two aggregates in one transaction, it's a signal that your aggregate boundaries may be wrong, or that you need to use domain events to coordinate between them.

class Order:  # Aggregate Root
    def __init__(self, order_id: str, customer_id: str):
        self.order_id = order_id
        self.customer_id = customer_id
        self._items = []          # Internal — not exposed directly
        self.status = "PENDING"

    def add_item(self, product_id: str, quantity: int, unit_price: Money):
        if self.status != "PENDING":
            raise ValueError("Cannot add items to a non-pending order")
        if quantity <= 0:
            raise ValueError("Quantity must be positive")
        # Business rule enforced here, not scattered across the codebase
        self._items.append(OrderItem(product_id, quantity, unit_price))

    def calculate_total(self) -> Money:
        return sum((item.subtotal() for item in self._items),
                   start=Money(0, "USD"))

class OrderItem:  # Internal entity — not accessible from outside the aggregate
    def __init__(self, product_id: str, quantity: int, unit_price: Money):
        self.product_id = product_id
        self.quantity = quantity
        self.unit_price = unit_price

    def subtotal(self) -> Money:
        return Money(self.unit_price.amount * self.quantity,
                     self.unit_price.currency)

Sizing Aggregates Correctly

A common mistake is making aggregates too large. If your Order aggregate contains the entire Customer object (with all their orders, addresses, and payment methods), every operation on an order locks the entire customer record. This kills concurrency and performance.

The rule of thumb: keep aggregates as small as possible. Reference other aggregates by ID only, not by object reference. If you need data from another aggregate, look it up separately.