Chapter 10: Domain Services — Cross-Aggregate Logic

"Place an order" is not just about the Order aggregate. It also involves checking inventory, reserving stock, and possibly more. When business logic spans multiple aggregates, a domain service encapsulates it.

When Logic Spans Aggregates

An aggregate's methods operate on its own state. But some business operations involve multiple aggregates:

• Order fulfillment: check Inventory stock → reserve items → confirm Order
• Transfer between accounts: debit one Account → credit another

These operations cannot live in a single aggregate. A domain service coordinates them while keeping business rules explicit.

Building the OrderFulfillmentService

    def __init__(self, order, inventories):
        super(OrderFulfillmentService, self).__init__(order, *inventories)
        self.order = order
        self.inventories = inventories

    def fulfill(self):
        """Check stock, reserve items, and confirm the order."""
        for item in self.order.items:
            inventory = next(
                (i for i in self.inventories if i.title == item.book_title),
                None,
            )
            inventory.reserve_stock(item.quantity)

        self.order.confirm()

    @invariant.pre
    def all_items_must_be_in_stock(self):
        for item in self.order.items:
            inventory = next(
                (i for i in self.inventories if i.title == item.book_title),
                None,
            )
            if inventory is None or inventory.quantity < item.quantity:
                raise ValidationError(
                    {"_service": [f"'{item.book_title}' is out of stock"]}
                )

    @invariant.pre
    def order_must_have_payment(self):
        if not self.order.payment_id:
            raise ValidationError(
                {"_service": ["Order must have a valid payment method"]}
            )

Key points:

• @domain.domain_service(part_of=[Order, Inventory]) declares that this service spans two aggregates. The part_of must list two or more aggregates.
• The constructor receives aggregate instances and calls super().__init__ with them.
• The fulfill() method orchestrates the business process: reserve stock for each item, then confirm the order.

Invariants on Domain Services

Domain services support the same @invariant.pre and @invariant.post decorators as aggregates:

@invariant.pre
def all_items_must_be_in_stock(self):
    for item in self.order.items:
        inventory = next(
            (i for i in self.inventories if i.title == item.book_title),
            None,
        )
        if inventory is None or inventory.quantity < item.quantity:
            raise ValidationError(
                {"_service": [f"'{item.book_title}' is out of stock"]}
            )

@invariant.pre
def order_must_have_payment(self):
    if not self.order.payment_id:
        raise ValidationError(
            {"_service": ["Order must have a valid payment method"]}
        )

• Pre-invariants run before fulfill() — they prevent the operation if preconditions are not met.
• Post-invariants (not shown here) would validate the final state after the operation.

Invariants run automatically when any method on the service is called. They are checked in the order they are defined.

Using the Domain Service

domain.init(traverse=False)


if __name__ == "__main__":
    with domain.domain_context():
        order_repo = domain.repository_for(Order)
        inventory_repo = domain.repository_for(Inventory)

        # Set up inventory
        inv1 = Inventory(book_id="book-1", title="The Great Gatsby", quantity=10)
        inv2 = Inventory(book_id="book-2", title="Brave New World", quantity=5)
        inventory_repo.add(inv1)
        inventory_repo.add(inv2)

        # Create an order
        order = Order(
            customer_name="Alice Johnson",
            payment_id="pay-123",
            items=[
                OrderItem(
                    book_title="The Great Gatsby",
                    quantity=2,
                    unit_price=Money(amount=12.99),
                ),
                OrderItem(
                    book_title="Brave New World",
                    quantity=1,
                    unit_price=Money(amount=14.99),
                ),
            ],
        )

        # Fulfill the order using the domain service
        print("=== Fulfilling Order ===")
        service = OrderFulfillmentService(order, [inv1, inv2])
        service.fulfill()

        print(f"Order status: {order.status}")
        print(f"Gatsby stock: {inv1.quantity}")
        print(f"Brave New World stock: {inv2.quantity}")

Run it:

$ python bookshelf.py
=== Fulfilling Order ===
Order status: CONFIRMED
Gatsby stock: 8
Brave New World stock: 4

=== Out of Stock Scenario ===
Caught: {'_service': ["'Brave New World' is out of stock"]}

=== Missing Payment Scenario ===
Caught: {'_service': ['Order must have a valid payment method']}

All checks passed!

The service enforces business rules before making any changes. If stock is insufficient or payment is missing, nothing happens — no partial updates.

Domain Services vs Application Services

These two types of services have different purposes:

Domain Services	Application Services
Contain business rules	Contain coordination logic
Span multiple aggregates	Work with one aggregate
Have invariants (pre/post)	Have @use_case for transactions
Called by handlers or app services	Called by API/UI layer
Part of the domain model	Part of the application layer

A typical flow:

graph LR
    API[API Endpoint] --> AS[Application Service]
    AS --> DS[Domain Service]
    DS --> A1[Order Aggregate]
    DS --> A2[Inventory Aggregate]

    style DS fill:#e8f5e9
    style AS fill:#e1f5fe

The application service receives the request, loads the aggregates, and passes them to the domain service. The domain service enforces rules and orchestrates the changes.

Full Source

from enum import Enum

from protean import Domain, invariant
from protean.exceptions import ValidationError
from protean.fields import Float, HasMany, Identifier, Integer, String, ValueObject

domain = Domain()


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


class OrderStatus(Enum):
    PENDING = "PENDING"
    CONFIRMED = "CONFIRMED"
    SHIPPED = "SHIPPED"


@domain.event(part_of="Order")
class OrderConfirmed:
    order_id = Identifier(required=True)
    customer_name = String(max_length=150, required=True)


@domain.aggregate
class Order:
    customer_name = String(max_length=150, required=True)
    payment_id = Identifier()
    status = String(
        max_length=20, choices=OrderStatus, default=OrderStatus.PENDING.value
    )
    items = HasMany("OrderItem")

    def confirm(self):
        self.status = OrderStatus.CONFIRMED.value
        self.raise_(OrderConfirmed(order_id=self.id, customer_name=self.customer_name))


@domain.entity(part_of=Order)
class OrderItem:
    book_title = String(max_length=200, required=True)
    quantity = Integer(required=True)
    unit_price = ValueObject(Money)


@domain.event(part_of="Inventory")
class StockReserved:
    book_id = Identifier(required=True)
    quantity = Integer(required=True)


@domain.aggregate
class Inventory:
    book_id = Identifier(required=True)
    title = String(max_length=200, required=True)
    quantity = Integer(default=0)

    def reserve_stock(self, amount: int):
        self.quantity -= amount
        self.raise_(StockReserved(book_id=self.book_id, quantity=amount))


@domain.domain_service(part_of=[Order, Inventory])
class OrderFulfillmentService:
    def __init__(self, order, inventories):
        super(OrderFulfillmentService, self).__init__(order, *inventories)
        self.order = order
        self.inventories = inventories

    def fulfill(self):
        """Check stock, reserve items, and confirm the order."""
        for item in self.order.items:
            inventory = next(
                (i for i in self.inventories if i.title == item.book_title),
                None,
            )
            inventory.reserve_stock(item.quantity)

        self.order.confirm()

    @invariant.pre
    def all_items_must_be_in_stock(self):
        for item in self.order.items:
            inventory = next(
                (i for i in self.inventories if i.title == item.book_title),
                None,
            )
            if inventory is None or inventory.quantity < item.quantity:
                raise ValidationError(
                    {"_service": [f"'{item.book_title}' is out of stock"]}
                )

    @invariant.pre
    def order_must_have_payment(self):
        if not self.order.payment_id:
            raise ValidationError(
                {"_service": ["Order must have a valid payment method"]}
            )




domain.init(traverse=False)


if __name__ == "__main__":
    with domain.domain_context():
        order_repo = domain.repository_for(Order)
        inventory_repo = domain.repository_for(Inventory)

        # Set up inventory
        inv1 = Inventory(book_id="book-1", title="The Great Gatsby", quantity=10)
        inv2 = Inventory(book_id="book-2", title="Brave New World", quantity=5)
        inventory_repo.add(inv1)
        inventory_repo.add(inv2)

        # Create an order
        order = Order(
            customer_name="Alice Johnson",
            payment_id="pay-123",
            items=[
                OrderItem(
                    book_title="The Great Gatsby",
                    quantity=2,
                    unit_price=Money(amount=12.99),
                ),
                OrderItem(
                    book_title="Brave New World",
                    quantity=1,
                    unit_price=Money(amount=14.99),
                ),
            ],
        )

        # Fulfill the order using the domain service
        print("=== Fulfilling Order ===")
        service = OrderFulfillmentService(order, [inv1, inv2])
        service.fulfill()

        print(f"Order status: {order.status}")
        print(f"Gatsby stock: {inv1.quantity}")
        print(f"Brave New World stock: {inv2.quantity}")

        # --- Pre-invariant: out of stock ---
        print("\n=== Out of Stock Scenario ===")
        big_order = Order(
            customer_name="Bob Smith",
            payment_id="pay-456",
            items=[
                OrderItem(
                    book_title="Brave New World",
                    quantity=100,  # More than available
                    unit_price=Money(amount=14.99),
                ),
            ],
        )
        try:
            OrderFulfillmentService(big_order, [inv2]).fulfill()
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # --- Pre-invariant: no payment ---
        print("\n=== Missing Payment Scenario ===")
        no_pay_order = Order(
            customer_name="Charlie",
            items=[
                OrderItem(
                    book_title="The Great Gatsby",
                    quantity=1,
                    unit_price=Money(amount=12.99),
                ),
            ],
        )
        try:
            OrderFulfillmentService(no_pay_order, [inv1]).fulfill()
        except ValidationError as e:
            print(f"Caught: {e.messages}")

        # Verify
        assert order.status == OrderStatus.CONFIRMED.value
        assert inv1.quantity == 8  # 10 - 2
        assert inv2.quantity == 4  # 5 - 1
        print("\nAll checks passed!")

Summary

In this chapter you learned:

• Domain services encapsulate business logic that spans multiple aggregates.
• @domain.domain_service(part_of=[Agg1, Agg2]) declares the aggregates involved.
• Domain services support invariants — pre-conditions and post-conditions that run automatically.
• Domain services are distinct from application services: they contain business rules, not coordination logic.

We now have all the building blocks for changing state. In the next chapter, we will build the read side — projections that provide optimized views of our data.

Next

Chapter 11: Projections and Projectors →