Design Small Aggregates

The problem

The most common structural mistake in Domain-Driven Design is an aggregate that is too big. Grouping related things feels right: a customer really does have orders, addresses, payment methods, and loyalty points, so they all go into one Customer.

That instinct gives you an aggregate that demos well and fails in production:

• Everything contends for one lock. Saving an aggregate writes the whole object graph as a unit, so changing a shipping address, adding a loyalty point, and placing an order all lock the same row. Two users touching unrelated parts of one customer queue behind each other for no reason.

• Every read pays for the whole graph. Changing a customer's email loads their orders, addresses, payment methods, and loyalty history first. You transfer, deserialize, and hold in memory the whole aggregate, even when the job needs only a small part of it.

• Transactions grow with it. More data in a transaction means more chance of a conflict, a lock held longer, and more work thrown away on rollback. In an event-sourced system it also means a long stream that is slow to replay.

• You end up enforcing invariants nobody asked for. Data belongs in an aggregate only when it has to stay consistent with the rest of the aggregate. Most relationships do not. Does an email change have to be atomic with the order history? Almost never. A big aggregate guarantees consistency the business never wanted.

• It sets hard. Once other code depends on the shape, splitting it is a large piece of work. By then, commands, events, handlers, and projections all depend on the aggregate's internal structure, so you have to migrate all of them to move one boundary.

All five problems have one cause: you drew the boundary around what relates to what, but it should follow what has to stay consistent.

---

The pattern

Design aggregates around consistency boundaries, not data relationships.

Keep in an aggregate only the data that has to change together, in one transaction, to hold a business rule. Everything else moves out: into its own aggregate referenced by identity, or into a value object where a copy of the data at that moment is what you want.

Wrong mental model:
  "These things are related, so they belong in the same aggregate."

Right mental model:
  "These things must be consistent with each other in the same transaction,
   so they belong in the same aggregate."

The consistency boundary test

Before you add anything to an aggregate, ask:

"If this data changes, must it be atomically consistent with the rest of the aggregate in the same transaction?"

If the answer is no, it belongs somewhere else.

Take an e-commerce system:

Concept	Must be consistent with Order in same transaction?	Belongs in Order aggregate?
Order line items	Yes, adding an item must update the total	Yes
Order status	Yes, status transitions have invariants	Yes
Customer profile	No, customer name can change independently	No
Shipping address	Depends, immutable snapshot at order time?	Maybe (as a value object)
Inventory count	No, inventory is a separate concern	No
Payment record	No, payment is processed asynchronously	No
Product catalog	No, product details are reference data	No

Order keeps an ID, its line items, a status, a total, and perhaps a snapshot of the shipping address as it was at the time. Everything else is its own aggregate, referenced by identity.

---

Reference by identity

When one aggregate needs to know about another, store its identity, never the object. This is what lets you keep aggregates small and still model the relationships between them.

The anti-pattern: embedding aggregates

# fragment
# Anti-pattern: Customer embedded inside Order
@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer = HasOne(Customer)        # Embeds the entire Customer aggregate
    items = HasMany(OrderItem)
    status: String(default="pending")
    total: Float(default=0.0)

Which gives you:

• Loading an Order loads the whole Customer with it
• Changing the Customer's email means loading the Order
• Order and Customer share one transaction boundary
• You cannot scale their storage separately

The pattern: reference by identity

# Pattern: Order references Customer by identity
@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer_id: Identifier(required=True)  # Just the identity
    items = HasMany("OrderItem")
    status: String(default="pending")
    total: Float(default=0.0)

Now the Order knows which customer placed it without owning or embedding them. The Identifier field holds the identity, a string, integer, or UUID, and nothing more.

When you need data, not the aggregate

Sometimes a handler needs data from another aggregate to decide something. The instinct is to embed it. You have three better options.

Option 1: Include the data in the command.

The caller already holds the data, so it goes in the command:

@domain.command(part_of=Order)
class PlaceOrder:
    order_id: Identifier(identifier=True)
    customer_id: Identifier(required=True)
    customer_name: String(required=True)  # Included by the caller
    customer_email: String(required=True)  # Included by the caller
    items: List(required=True)

The handler never loads the Customer. The command carries everything the Order needs.

Option 2: Store a snapshot as a value object.

When you want another aggregate's data frozen as it was:

@domain.value_object
class CustomerSnapshot:
    customer_id: String(required=True)
    name: String(required=True)
    email: String(required=True)


@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer = ValueObject(CustomerSnapshot)  # Snapshot, not the live aggregate
    items = HasMany("OrderItem")
    status: String(default="pending")
    total: Float(default=0.0)

CustomerSnapshot is a value object: immutable, embedded, and holding the customer's details as of the moment the order was placed. Later profile edits do not touch it.

Option 3: Look up in a read model.

When your handler needs the other aggregate's data to decide, and a slightly out-of-date copy is acceptable, query a projection:

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder):
        # Check customer's credit status via a read model
        customer_view = current_domain.repository_for(CustomerCreditView).get(
            command.customer_id
        )

        if customer_view.credit_status == "suspended":
            raise ValidationError({"customer_id": ["Customer credit is suspended"]})

        order = Order(
            order_id=command.order_id,
            customer_id=command.customer_id,
            items=command.items,
        )
        current_domain.repository_for(Order).add(order)

You read what you need without tying Order to Customer. A projector runs after the event it reacts to, so the view can trail the aggregate it reflects. Query a projection only where that delay is acceptable. Where the delay is not acceptable, make the decision inside the other aggregate.

---

What Protean gives you

Protean gives you four pieces for keeping aggregates small:

The Identifier field

Identifier exists for exactly this. It holds another aggregate's identity without embedding it:

@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer_id: Identifier(required=True)  # References Customer aggregate
    product_id: Identifier(required=True)  # References Product aggregate


@domain.aggregate
class Shipment:
    shipment_id: Auto(identifier=True)
    order_id: Identifier(required=True)  # References Order aggregate
    carrier_id: Identifier(required=True)  # References Carrier aggregate

Each aggregate stands alone, and the relationships between them are identities.

Entities for real composition

When data really does belong inside the aggregate, because it has to be consistent in the same transaction, use entities:

@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer_id: Identifier(required=True)
    items = HasMany("OrderItem")
    status: String(default="draft")

    def add_item(self, product_id, product_name, quantity, unit_price):
        item = OrderItem(
            product_id=product_id,
            product_name=product_name,
            quantity=quantity,
            unit_price=unit_price,
        )
        self.add_items(item)

    @invariant.post
    def order_must_have_items(self):
        if self.status != "draft" and not self.items:
            raise ValidationError({"items": ["Order must have at least one item"]})


@domain.entity(part_of=Order)
class OrderItem:
    product_id: Identifier(required=True)
    product_name: String(required=True)
    quantity: Integer(min_value=1, required=True)
    unit_price: Float(required=True)

Order contains OrderItem because:

• Adding or removing one has to update the order's total in the same transaction
• The order has rules about its items, such as needing at least one
• An item does not exist without its order

Value objects for embedded data

When data describes part of the aggregate but has no identity of its own:

@domain.value_object
class Money:
    amount: Float(required=True)
    currency: String(max_length=3, required=True)


@domain.value_object
class ShippingAddress:
    street: String(required=True)
    city: String(required=True)
    state: String(required=True)
    postal_code: String(required=True)
    country: String(required=True)


@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer_id: Identifier(required=True)
    items = HasMany("OrderItem")
    total = ValueObject(Money)
    shipping_address = ValueObject(ShippingAddress)  # Snapshot at order time

ShippingAddress is embedded as a value object, a frozen record of where this order goes. It does not point at an Address aggregate, because the order needs the address as it was when the order was placed, even after the customer changes it.

Domain events across aggregates

When something in one aggregate has to change another, raise a domain event:

@domain.event(part_of="Order")
class OrderPlaced:
    order_id: Identifier(required=True)
    customer_id: Identifier(required=True)
    total_amount: Float(required=True)
    items: List(required=True)  # Each item: product_id, quantity


@domain.aggregate
class Order:
    order_id: Auto(identifier=True)
    customer_id: Identifier(required=True)
    items = HasMany("OrderItem")
    status: String(default="draft")
    total = ValueObject(Money)

    def place(self):
        if self.status != "draft":
            raise ValidationError({"status": ["Only draft orders can be placed"]})

        self.status = "placed"
        self.raise_(
            OrderPlaced(
                order_id=self.order_id,
                customer_id=self.customer_id,
                total_amount=self.total.amount,
                items=[
                    {"product_id": item.product_id, "quantity": item.quantity}
                    for item in self.items
                ],
            )
        )


# A separate aggregate, changed only by its own command handler
@domain.aggregate
class CustomerLoyalty:
    customer_id: Identifier(identifier=True)
    points: Integer(default=0)
    applied_order_ids: List(content_type=String)

    def add_points(self, order_id, points):
        self.points += points
        self.applied_order_ids = [*self.applied_order_ids, order_id]


@domain.command(part_of=CustomerLoyalty)
class AwardPoints:
    order_id: Identifier(required=True)
    customer_id: Identifier(required=True)
    points: Integer(required=True)


@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        current_domain.process(
            AwardPoints(
                order_id=event.order_id,
                customer_id=event.customer_id,
                points=int(event.total_amount),
            )
        )


@domain.command_handler(part_of=CustomerLoyalty)
class LoyaltyCommandHandler:
    @handle(AwardPoints)
    def award_points(self, command: AwardPoints):
        repo = current_domain.repository_for(CustomerLoyalty)
        loyalty = repo.get(command.customer_id)
        if command.order_id in loyalty.applied_order_ids:
            return  # This order was already applied
        loyalty.add_points(command.order_id, command.points)
        repo.add(loyalty)

Order knows nothing about CustomerLoyalty. It raises an event. The event handler stays with Order, the aggregate that owns the event, and issues an AwardPoints command. CustomerLoyalty's own command handler updates the points. You can deploy, scale, and test Order and CustomerLoyalty on their own.

Events are delivered at least once, so the same OrderPlaced can arrive twice. The command carries the order id, and the loyalty command handler returns without changes when it has already applied that order. The event handlers guide covers this hand-off in detail.

---

The two-aggregate rule

A rule of thumb: if an operation looks like it has to change two aggregates, it probably wants an event instead.

When you catch yourself loading and changing two aggregates in one handler:

# fragment
# Anti-pattern: modifying two aggregates in one handler
@handle(PlaceOrder)
def place_order(self, command: PlaceOrder):
    order_repo = current_domain.repository_for(Order)
    order = Order(order_id=command.order_id, items=command.items)
    order.place()
    order_repo.add(order)

    # This should NOT be here
    inventory_repo = current_domain.repository_for(Inventory)
    for item in command.items:
        inventory = inventory_repo.get(item.product_id)
        inventory.reserve(item.quantity)
        inventory_repo.add(inventory)

Split it. The command handler changes only the Order. An event handler on Order reacts to OrderPlaced by issuing a ReserveStock command for each product, and Inventory's command handler does the write:

# Pattern: one aggregate per handler, an event and a command for the rest
@domain.command(part_of=Order)
class PlaceOrder:
    order_id: Identifier(identifier=True)
    items: List(required=True)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command: PlaceOrder):
        order_repo = current_domain.repository_for(Order)
        order = Order(
            order_id=command.order_id,
            items=command.items,
        )
        order.place()  # Raises OrderPlaced event
        order_repo.add(order)


@domain.aggregate
class Inventory:
    product_id: Identifier(identifier=True)
    available: Integer(default=0)
    reserved: Integer(default=0)
    applied_order_ids: List(content_type=String)

    def reserve(self, order_id, quantity):
        self.available -= quantity
        self.reserved += quantity
        self.applied_order_ids = [*self.applied_order_ids, order_id]


@domain.command(part_of=Inventory)
class ReserveStock:
    order_id: Identifier(required=True)
    product_id: Identifier(required=True)
    quantity: Integer(required=True)


@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderPlaced)
    def on_order_placed(self, event: OrderPlaced):
        # One ReserveStock per product, so the order id alone marks it applied
        quantities = {}
        for item in event.items:
            product_id = item["product_id"]
            quantities[product_id] = quantities.get(product_id, 0) + item["quantity"]

        for product_id, quantity in quantities.items():
            current_domain.process(
                ReserveStock(
                    order_id=event.order_id,
                    product_id=product_id,
                    quantity=quantity,
                )
            )


@domain.command_handler(part_of=Inventory)
class InventoryCommandHandler:
    @handle(ReserveStock)
    def reserve_stock(self, command: ReserveStock):
        repo = current_domain.repository_for(Inventory)
        inventory = repo.get(command.product_id)
        if command.order_id in inventory.applied_order_ids:
            return  # This order was already applied
        inventory.reserve(command.order_id, command.quantity)
        repo.add(inventory)

A well-designed DDD system has this shape: small aggregates, connected by events and by the commands that event handlers issue.

---

A worked example

Take a project management system with these requirements:

• A Project has a name, description, and status
• A Team is assigned to a project and has members
• Tasks belong to a project and are assigned to team members
• Comments are posted on tasks
• Time entries are logged against tasks
• When a task is completed, the project's progress percentage updates

The naive design

# fragment
# Anti-pattern: everything in one aggregate
@domain.aggregate
class Project:
    name: String(required=True)
    description: Text()
    status: String(default="active")
    team = HasOne(Team)
    tasks = HasMany(Task)              # Could be hundreds
    time_entries = HasMany(TimeEntry)   # Could be thousands
    progress: Float(default=0.0)

Logging a single time entry loads and saves the whole Project, with every task, team member, comment, and time entry attached. On a project with 500 tasks and 2,000 time entries that is unusable.

The refactored design

Put each relationship through the consistency boundary test:

Concept	Must be atomically consistent with Project?	Decision
Project name/status	Yes (it IS the project)	Inside Project
Team	No; team changes are independent	Separate aggregate
Task	No, tasks change independently	Separate aggregate
Task comments	No, comments are independent	Inside Task (entity)
Time entries	No, logging time is independent	Separate aggregate
Progress	Derived, can be eventually consistent	Updated via events

@domain.aggregate
class Project:
    project_id: Auto(identifier=True)
    name: String(required=True)
    description: Text()
    status: String(default="active")
    progress: Float(default=0.0)

    def update_progress(self, completed_count, total_count):
        if total_count > 0:
            self.progress = (completed_count / total_count) * 100


@domain.aggregate
class Team:
    team_id: Auto(identifier=True)
    project_id: Identifier(required=True)  # References Project
    members = HasMany("TeamMember")


@domain.entity(part_of=Team)
class TeamMember:
    user_id: Identifier(required=True)
    role: String(default="member")


@domain.aggregate
class Task:
    task_id: Auto(identifier=True)
    project_id: Identifier(required=True)  # References Project
    assignee_id: Identifier()  # References a User
    title: String(required=True)
    status: String(default="open")
    comments = HasMany("Comment")

    def complete(self):
        if self.status == "completed":
            return
        self.status = "completed"
        self.raise_(
            TaskCompleted(
                task_id=self.task_id,
                project_id=self.project_id,
            )
        )


@domain.event(part_of=Task)
class TaskCompleted:
    task_id: Identifier(required=True)
    project_id: Identifier(required=True)


@domain.entity(part_of=Task)
class Comment:
    author_id: Identifier(required=True)
    content: Text(required=True)
    posted_at: DateTime()


@domain.aggregate
class TimeEntry:
    entry_id: Auto(identifier=True)
    task_id: Identifier(required=True)  # References Task
    user_id: Identifier(required=True)  # References User
    hours: Float(required=True)
    description: Text()

Each aggregate is now small:

• Project: Just name, status, progress (updated eventually via events)
• Team: Members for a project (changes independently of project)
• Task: Title, status, comments (comments are entities because they belong to the task's consistency boundary)
• TimeEntry: Standalone records (no invariant ties them to the task's state)

Completing a task raises an event. The event handler on Task issues a RecalculateProgress command, and Project's command handler recounts the tasks:

@domain.command(part_of=Project)
class RecalculateProgress:
    project_id: Identifier(required=True)


@domain.event_handler(part_of=Task)
class TaskEventHandler:
    @handle(TaskCompleted)
    def on_task_completed(self, event: TaskCompleted):
        current_domain.process(RecalculateProgress(project_id=event.project_id))


@domain.command_handler(part_of=Project)
class ProjectCommandHandler:
    @handle(RecalculateProgress)
    def recalculate_progress(self, command: RecalculateProgress):
        # `count()` issues a SELECT COUNT(*); it does not load the tasks
        tasks = current_domain.repository_for(Task).query.filter(
            project_id=command.project_id
        )
        repo = current_domain.repository_for(Project)
        project = repo.get(command.project_id)
        project.update_progress(
            tasks.filter(status="completed").count(),
            tasks.count(),
        )
        repo.add(project)

The command handler counts the rows itself, so running it twice for the same event gives the same progress. Do not take those counts from a projection that a projector updates from TaskCompleted as well. The event handler on Task and the projector consume the event through separate subscriptions, in no fixed order. Adding one to the projection's count double-counts the task when the projector ran first. The same addition gives a stale count when the projector has not run yet.

---

When to do something else

Data that really must be consistent

Sometimes data does have to share an aggregate. An invoice and its line items always have to agree, because the total has to match the sum of the lines. The debit and credit of a transfer within one account have to be atomic. Leave those alone.

The test does not change: must these move together, in one transaction? If yes, they stay together.

Small domains

In a small application with a handful of entities and almost no concurrency, splitting into many aggregates may cost more than it returns. With one user at a time and a few hundred records, a larger aggregate that keeps the code short is a fair trade.

Start with the larger aggregate. Split it when contention, performance, or complexity becomes a problem.

Event-sourced aggregates

Event-sourced aggregates carry one more concern: stream length. A long-lived aggregate with thousands of events takes longer to replay, which is a reason to keep event-sourced aggregates small. Where an aggregate genuinely needs to be one consistency boundary, reach for snapshots before you split it. Once the events since the last snapshot reach snapshot_threshold, Protean writes a snapshot during the next load of the aggregate. You can also create a snapshot by hand (see Snapshots).

---

Common mistakes

Mistake 1: splitting on UI screens

"The order page shows orders and the customer page shows customer info, so they should be separate aggregates." That reasoning gives the right split for the wrong reason. Boundaries come from business rules, not screen layout. Put orders and customers on one page tomorrow and the boundaries should not move.

Mistake 2: using HasOne/HasMany across aggregates

# fragment
# Mistake: using association fields for separate aggregates
@domain.aggregate
class Order:
    customer = HasOne(Customer)  # This embeds Customer inside Order

HasOne and HasMany are for entities inside the aggregate. To point at another aggregate, use Identifier:

# Correct: identity reference to another aggregate
@domain.aggregate
class Order:
    customer_id: Identifier(required=True)  # References Customer

Mistake 3: splitting too early

A long field list is not a reason to split. A User with 15 fields covering name, email, phone, preferences, and settings is fine when those fields have to agree with each other. Count of fields is not the test. Consistency is.

---

Summary

Aspect	Large Aggregates	Small Aggregates
Boundary criterion	Data relationships	Consistency requirements
Cross-aggregate reference	Embed (HasOne/HasMany)	Identity (Identifier)
Cross-aggregate changes	Same transaction	Domain events
Loading cost	Entire object graph	Only what's needed
Concurrency	High contention	Low contention
Transaction scope	Large (risky)	Small (safe)
Scalability	Limited	Independent per aggregate
Testability	Requires full graph	Isolated units

Draw the boundary around what has to stay consistent. Point at other aggregates by identity, and let domain events carry the work between them.

---

Related reading

Concepts:

• Aggregates: What aggregates are and their core properties.
• Entities: Objects with identity within aggregates.

Guides:

• Aggregates: Defining aggregates, fields, and configuration.
• Relationships: Connecting entities within aggregate boundaries.
• Choosing Element Types: Choosing between aggregates, entities, and value objects.