Chapter 13: Checking Before You Ship with Domain Services

A customer just complained: they ordered five copies of a book, the order was confirmed, but only three were in stock. The current system confirms orders blindly. It never checks inventory. We need business logic that spans two aggregates (Order and Inventory), and that logic does not belong in either aggregate. It belongs in a domain service.

What Is a Domain Service?

A domain service is a stateless object that encapsulates business logic spanning two or more aggregates. Unlike aggregates, domain services:

• Have no identity and no lifecycle.
• Are invoked from command handlers or application services.
• Can run pre-invariants and post-invariants for validation.
• Are always associated with the aggregates they coordinate.

When to Use a Domain Service (vs. an Event Handler)

You might wonder: couldn't we use an event handler to check inventory when an order is confirmed? The key difference is transactional consistency:

Approach	Guarantees
Domain service	Synchronous. Inventory is checked before the order is confirmed. If the stock it reads is insufficient, the command fails and the order stays PENDING.
Event handler	Eventually consistent; the order is confirmed first, then the handler runs. If stock is insufficient, you need compensating actions.

Use a domain service when the business rule says "this must not happen", like confirming an order without sufficient stock. Use event handlers when the reaction can happen after the fact.

This chapter uses both. The domain service checks stock and confirms the order. Reserving the stock changes a second aggregate, Inventory, so it happens in its own transaction, through an event and a command.

The Fulfillment Service

@domain.domain_service(part_of=[Order, Inventory])
class OrderFulfillmentService:
    """Validates inventory availability before confirming an order."""

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

    @invariant.pre
    def all_items_in_stock(self):
        """Check that every order item has sufficient inventory."""
        inventory_by_title = {inv.title: inv for inv in self.inventories}

        for title, quantity in self.order.quantities_by_title().items():
            inv = inventory_by_title.get(title)
            if inv is None:
                raise ValidationError(
                    {"_entity": [f"No inventory record for '{title}'"]}
                )
            if inv.quantity < quantity:
                raise ValidationError(
                    {
                        "_entity": [
                            (
                                f"Insufficient stock for '{title}': "
                                f"{inv.quantity} available, {quantity} requested"
                            )
                        ]
                    }
                )

    def confirm_order(self):
        """Confirm the order. The pre-invariant has already checked stock."""
        self.order.confirm()
        return self.order

Let's break down how this works:

1. part_of=[Order, Inventory]: The service is associated with both aggregates. Protean requires domain services to declare which aggregates they coordinate.

2. __init__: The constructor receives the aggregates and calls super().__init__() with all of them, which keeps them on the service. Protean wires the invariants when it registers the class, so they run even if a service leaves this call out.

3. @invariant.pre: The all_items_in_stock invariant runs before confirm_order() executes. If any item is out of stock, a ValidationError is raised and the order is never confirmed. The check adds up the order's lines for each title, so two lines for the same book are checked against the stock together. Pre-invariants are the domain service's main value. They check a rule that spans aggregates before anything changes.

4. confirm_order(): The domain method that performs the mutation. It confirms the order, and Order.confirm() raises OrderConfirmed with the customer's name and the quantity ordered for each title. The service does not touch Inventory.

Updating the Command Handler

The ConfirmOrder handler now loads inventory records and delegates to the domain service:

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(ConfirmOrder)
    def confirm_order(self, command: ConfirmOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        if order.status == "CONFIRMED":
            # A repeated ConfirmOrder would raise OrderConfirmed a second time.
            return

        # Load inventory records for all items in the order
        inv_repo = current_domain.repository_for(Inventory)
        inventories = []
        for title in order.quantities_by_title():
            inv_results = inv_repo.query.filter(title=title).all()
            if inv_results.items:
                inventories.append(inv_results.items[0])

        # Delegate to the domain service
        service = OrderFulfillmentService(order, inventories)
        service.confirm_order()

        # Persist only the order. Inventory reserves through OrderConfirmed.
        repo.add(order)

The handler's job is orchestration, not business logic:

1. Load the order from the repository. Stop if it is already confirmed, so a repeated ConfirmOrder changes nothing.
2. Load the relevant inventory records.
3. Pass everything to the domain service.
4. Persist the order. Only the order.

If the inventory check fails, the ValidationError propagates to the API layer and returns a 400 Bad Request automatically (thanks to the exception handlers we registered in Chapter 10). Nothing was saved, so the order stays PENDING.

A handler changes one aggregate per transaction. The inventory records are loaded so the service can check them, but they are not saved here.

The handlers find a book's stock by its title, so the tutorial assumes each title has one inventory record. If two records share a title, the reservation below fails with TooManyObjectsError after the order is saved.

Reserving the Stock

OrderConfirmed lists each title once, with its total quantity. An event handler in Order's cluster reacts to it and sends one ReserveInventory command per title. Inventory's command handler does the reservation:

@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderConfirmed)
    def on_order_confirmed(self, event: OrderConfirmed) -> None:
        print(
            f"  [Notification] Order {event.order_id} confirmed for {event.customer_name}"
        )

    @handle(OrderConfirmed)
    def reserve_inventory(self, event: OrderConfirmed) -> None:
        for item in event.items:
            current_domain.process(
                ReserveInventory(
                    order_id=event.order_id,
                    book_title=item["book_title"],
                    quantity=item["quantity"],
                )
            )


@domain.command_handler(part_of=Inventory)
class InventoryCommandHandler:
    @handle(ReserveInventory)
    def reserve(self, command: ReserveInventory) -> None:
        repo = current_domain.repository_for(Inventory)
        inventory = repo.find_by(title=command.book_title)
        # OrderConfirmed lists each title once, so an order reserves a given
        # book once. A redelivered event finds its order here and stops.
        if command.order_id in inventory.reserved_order_ids:
            return
        inventory.reserve(command.order_id, command.quantity)
        repo.add(inventory)

The event handler sits in Order's cluster because Order owns OrderConfirmed. It does not change Inventory itself. It hands the change to Inventory as a command. reserve_inventory belongs in the OrderEventHandler from Chapter 6, next to its on_order_confirmed notification. Add the method to that class; do not define a second OrderEventHandler. One handler can have several methods for the same event, and each one runs.

Events can be delivered more than once, so the command handler checks reserved_order_ids first and skips an order it has already reserved for. Without that check, a redelivered OrderConfirmed would reserve the stock twice. The check works because an order reserves each book once: two lines for the same book arrive as one command with the summed quantity. The list grows by one id for every order that reserves the book, and it loads with the inventory record each time.

The stock check and the reservation happen in two transactions. The service checks the stock it read. The reservation runs later, in its own transaction, and can still fail: another order can take the last copies in between. Inventory.reserve() checks the stock again and raises ValidationError, so the stock never goes below zero. The order, though, is already saved as CONFIRMED, and no stock is reserved for it. With the "sync" settings below, domain.process(ConfirmOrder(...)) then raises TransactionError. With asynchronous processing, the reservation fails in the server instead.

A real store has to handle that failure. A common way is a compensating step: Inventory's command handler catches the error and raises an event such as ReservationFailed, and a handler reacts by cancelling the order. Chapter 20 builds this kind of compensation with a process manager for a failed shipment. It leaves a failed reservation out, and describes how to add it.

This chapter's domain sets command_processing and event_processing to "sync", so the whole chain runs before domain.process() returns. With asynchronous processing, the reservation runs shortly after, in the server.

Testing the Domain Service

Test both the happy path and the out-of-stock scenario:

# tests/test_domain_services.py (example tests)


def test_confirm_order_with_stock():
    """Order is confirmed when inventory is sufficient."""
    with domain.domain_context():
        inv = Inventory(book_id="book-1", title="Dune", quantity=10)
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(book_title="Dune", quantity=2, unit_price=Money(amount=15.99))
            ],
        )

        service = OrderFulfillmentService(order, [inv])
        service.confirm_order()

        assert order.status == "CONFIRMED"
        assert inv.quantity == 10  # the service checks stock; reserving comes later


def test_confirm_order_out_of_stock():
    """Order fails when inventory is insufficient."""
    with domain.domain_context():
        inv = Inventory(book_id="book-1", title="Dune", quantity=1)
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(book_title="Dune", quantity=5, unit_price=Money(amount=15.99))
            ],
        )

        try:
            service = OrderFulfillmentService(order, [inv])
            service.confirm_order()
            raise AssertionError("Should have raised ValidationError")
        except ValidationError as e:
            assert "Insufficient stock" in str(e.messages)

Notice that we test the domain service directly, with no command processing and no repositories. The tests open a domain context because Order.confirm() raises an event. The service takes aggregates as input and changes them in memory, so it is easy to unit test. The happy-path test also shows that the service leaves the inventory alone: the stock is still 10 after the order is confirmed.

What We Built

• An OrderFulfillmentService domain service that validates inventory across aggregates before confirming an order.
• A @invariant.pre that enforces "all items must be in stock" as a cross-aggregate business rule.
• An updated ConfirmOrder handler that delegates to the service and persists only the order.
• An OrderConfirmed event, an event handler method that turns it into one ReserveInventory command per title, and an Inventory command handler that reserves the stock once per order.
• Tests verifying both success and failure paths.

In the next chapter, we will integrate with an external book supplier using a subscriber.

Full Source

from protean import Domain, handle, invariant
from protean.exceptions import ValidationError
from protean.fields import (
    Dict,
    Float,
    HasMany,
    Identifier,
    Integer,
    List,
    String,
    ValueObject,
)
from protean.utils.globals import current_domain

domain = Domain("bookshelf")
domain.config["command_processing"] = "sync"
domain.config["event_processing"] = "sync"


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


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

    def reserve(self, order_id: str, amount: int):
        if self.quantity < amount:
            raise ValidationError(
                {
                    "quantity": [
                        f"Insufficient stock: {self.quantity} available, {amount} requested"
                    ]
                }
            )
        self.quantity -= amount
        self.reserved_order_ids = [*self.reserved_order_ids, order_id]


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

    def quantities_by_title(self) -> dict[str, int]:
        """Total quantity per book, adding up lines for the same title."""
        totals: dict[str, int] = {}
        for item in self.items:
            totals[item.book_title] = totals.get(item.book_title, 0) + item.quantity
        return totals

    def confirm(self):
        self.status = "CONFIRMED"
        self.raise_(
            OrderConfirmed(
                order_id=self.id,
                customer_name=self.customer_name,
                items=[
                    {"book_title": title, "quantity": quantity}
                    for title, quantity in self.quantities_by_title().items()
                ],
            )
        )


@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=Order)
class OrderConfirmed:
    order_id: Identifier(required=True)
    customer_name: String(max_length=150, required=True)
    items = List(content_type=Dict)


@domain.domain_service(part_of=[Order, Inventory])
class OrderFulfillmentService:
    """Validates inventory availability before confirming an order."""

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

    @invariant.pre
    def all_items_in_stock(self):
        """Check that every order item has sufficient inventory."""
        inventory_by_title = {inv.title: inv for inv in self.inventories}

        for title, quantity in self.order.quantities_by_title().items():
            inv = inventory_by_title.get(title)
            if inv is None:
                raise ValidationError(
                    {"_entity": [f"No inventory record for '{title}'"]}
                )
            if inv.quantity < quantity:
                raise ValidationError(
                    {
                        "_entity": [
                            (
                                f"Insufficient stock for '{title}': "
                                f"{inv.quantity} available, {quantity} requested"
                            )
                        ]
                    }
                )

    def confirm_order(self):
        """Confirm the order. The pre-invariant has already checked stock."""
        self.order.confirm()
        return self.order




@domain.command(part_of=Order)
class ConfirmOrder:
    order_id: Identifier(required=True)


@domain.command(part_of=Inventory)
class ReserveInventory:
    order_id: Identifier(required=True)
    book_title: String(max_length=200, required=True)
    quantity: Integer(required=True)


@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(ConfirmOrder)
    def confirm_order(self, command: ConfirmOrder) -> None:
        repo = current_domain.repository_for(Order)
        order = repo.get(command.order_id)
        if order.status == "CONFIRMED":
            # A repeated ConfirmOrder would raise OrderConfirmed a second time.
            return

        # Load inventory records for all items in the order
        inv_repo = current_domain.repository_for(Inventory)
        inventories = []
        for title in order.quantities_by_title():
            inv_results = inv_repo.query.filter(title=title).all()
            if inv_results.items:
                inventories.append(inv_results.items[0])

        # Delegate to the domain service
        service = OrderFulfillmentService(order, inventories)
        service.confirm_order()

        # Persist only the order. Inventory reserves through OrderConfirmed.
        repo.add(order)




@domain.event_handler(part_of=Order)
class OrderEventHandler:
    @handle(OrderConfirmed)
    def on_order_confirmed(self, event: OrderConfirmed) -> None:
        print(
            f"  [Notification] Order {event.order_id} confirmed for {event.customer_name}"
        )

    @handle(OrderConfirmed)
    def reserve_inventory(self, event: OrderConfirmed) -> None:
        for item in event.items:
            current_domain.process(
                ReserveInventory(
                    order_id=event.order_id,
                    book_title=item["book_title"],
                    quantity=item["quantity"],
                )
            )


@domain.command_handler(part_of=Inventory)
class InventoryCommandHandler:
    @handle(ReserveInventory)
    def reserve(self, command: ReserveInventory) -> None:
        repo = current_domain.repository_for(Inventory)
        inventory = repo.find_by(title=command.book_title)
        # OrderConfirmed lists each title once, so an order reserves a given
        # book once. A redelivered event finds its order here and stops.
        if command.order_id in inventory.reserved_order_ids:
            return
        inventory.reserve(command.order_id, command.quantity)
        repo.add(inventory)




domain.init(traverse=False)


# tests/test_domain_services.py (example tests)


def test_confirm_order_with_stock():
    """Order is confirmed when inventory is sufficient."""
    with domain.domain_context():
        inv = Inventory(book_id="book-1", title="Dune", quantity=10)
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(book_title="Dune", quantity=2, unit_price=Money(amount=15.99))
            ],
        )

        service = OrderFulfillmentService(order, [inv])
        service.confirm_order()

        assert order.status == "CONFIRMED"
        assert inv.quantity == 10  # the service checks stock; reserving comes later


def test_confirm_order_out_of_stock():
    """Order fails when inventory is insufficient."""
    with domain.domain_context():
        inv = Inventory(book_id="book-1", title="Dune", quantity=1)
        order = Order(
            customer_name="Alice",
            items=[
                OrderItem(book_title="Dune", quantity=5, unit_price=Money(amount=15.99))
            ],
        )

        try:
            service = OrderFulfillmentService(order, [inv])
            service.confirm_order()
            raise AssertionError("Should have raised ValidationError")
        except ValidationError as e:
            assert "Insufficient stock" in str(e.messages)

Next

Chapter 14: Connecting to the Outside World with Subscribers →