Domain Services

DDD CQRS ES

Some business operations naturally span two or more aggregates. For example, placing an order requires confirming the order and reserving inventory. The two aggregates must be validated together. If you put this logic in the Order aggregate, it needs to know about Inventory; if you put it in the command handler, business rules leak into the application layer. Domain services solve this: they encapsulate cross-aggregate business logic in the domain layer, where it belongs.

For background on when and why to use domain services, see Domain Services concept.

Persist one aggregate, let events handle the rest

A domain service reads several aggregates to check its invariants, but it changes only one, the primary aggregate. The command handler invoking the service persists that aggregate. The primary aggregate raises an event, and an event handler issues a command that changes the other aggregate later. This preserves the one aggregate per transaction rule.

Defining a Domain Service

A Domain Service is defined with the Domain.domain_service decorator, and associated with at least two aggregates with the part_of option. The part_of option is required and must be a list of two or more aggregates, a domain service that operates on a single aggregate should be logic on the aggregate itself.

The service methods in a Domain Service can be structured in three flavors:

1. Class with class methods

If you don't have any invariants to be managed by the Domain Service, each method in the Domain Service can be a class method, that receives all the input necessary for performing the business function. The method checks the stock itself and returns the order.

# fragment
@domain.domain_service(part_of=[Order, Inventory])
class OrderPlacementService:
    @classmethod
    def place_order(cls, order, inventories):
        for product_id, quantity in order.quantities_by_product().items():
            inventory = next(
                (i for i in inventories if i.product_id == product_id), None
            )
            if inventory is None or inventory.quantity < quantity:
                raise ValidationError({"_service": ["Product is out of stock"]})

        order.confirm()
        return order

Invoking it is straightforward:

# fragment
OrderPlacementService.place_order(order, inventories)

Note

Protean does not run a Domain Service's invariants around its class methods, so a class-method service should not declare any. If you need pre/post invariants, use the instance-method or callable-class flavor. Protean still wraps a public class method and treats its first argument as the service instance. It runs that argument's own invariants around the call, so pass an aggregate first: a list there raises AttributeError. Call the method on the class. A call on an instance raises TypeError, because the instance is passed as one argument too many.

2. Class with instance methods

In this flavor, the Domain Service is instantiated with the aggregates and each method performs a distinct business function.

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

        self.order = order
        self.inventories = inventories

    def place_order(self):
        self.order.confirm()

    @invariant.pre
    def inventory_should_have_sufficient_stock(self):
        for product_id, quantity in self.order.quantities_by_product().items():
            inventory = next(
                (i for i in self.inventories if i.product_id == product_id), None
            )
            if inventory is None or inventory.quantity < quantity:
                raise ValidationError({"_service": ["Product is out of stock"]})

You would then instantiate the Domain Service, passing the relevant aggregates and invoke the methods on the instance.

# fragment
service = OrderPlacementService(order, inventories)
service.place_order()

3. Callable class

If you have a single business function, you can model it as a callable class:

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

        self.order = order
        self.inventories = inventories

    def __call__(self):
        self.order.confirm()

    @invariant.pre
    def inventory_should_have_sufficient_stock(self):
        for product_id, quantity in self.order.quantities_by_product().items():
            inventory = next(
                (i for i in self.inventories if i.product_id == product_id), None
            )
            if inventory is None or inventory.quantity < quantity:
                raise ValidationError({"_service": ["Product is out of stock"]})

# fragment
service = place_order(order, inventories)
service()

Deciding between flavors

Criterion	Class methods	Instance methods	Callable class
Invariants needed	No	Yes	Yes
Multiple operations	Yes	Yes	No (single `__call__`)
Simplest for single operation	—	—	Yes

The decision between instance methods and a callable class boils down to:

How many business functions does the Domain Service have? If only one, a callable class is the simpler shape.
Do you have pre invariants that only apply to specific methods? Then construct each method as a separate callable class. If invariants apply to all methods, a class with instance methods is preferable.

As your domain model matures, review regularly and decide on the best way to model the Domain Service.

Note

Invariants only wrap public methods and __call__. They skip dunder methods (other than __call__) and private methods (prefixed with _). A public method that calls another public method on the same instance runs the invariants again around the inner call. An invariant that calls a public method fails: that method's wrapper runs the invariant again, which calls the method again, until Python's recursion limit stops the call. Put helpers in private methods (prefixed with _) so they are not wrapped.

Domain Service vs Application Service vs Command Handler

These three constructs coordinate behavior, but at different levels:

Aspect	Domain Service	Application Service	Command Handler
Contains business logic	Yes, cross-aggregate rules	No, orchestration only	No, orchestration only
Operates on	2+ aggregates	1 aggregate	1 aggregate
Has invariants	Yes (`@invariant.pre`/`.post`)	No	No
Invoked by	Command handlers, app services	External callers (API, CLI)	`domain.process(command)`
Returns values	Optional	Yes (synchronous)	No (fire-and-forget)

For detailed guidance, see the Application Service vs Command Handler pattern.

Typical Workflow

Let us consider an example OrderPlacementService that checks inventory stock and confirms an order. The typical workflow of a Domain Service is below:

Sequence diagram

The handler loads the necessary aggregates through repositories, invokes the domain service to perform the cross-aggregate operation, and then persists only the primary aggregate. OrderConfirmed carries the ordered items. An event handler in the order's cluster reacts to it and issues a command to each inventory, and the inventory's command handler reserves the stock in its own transaction. Chapter 13 of the tutorial builds the whole flow, including the event handler and the inventory command.

Handler Integration Example

# fragment
@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def handle_place_order(self, command: PlaceOrder):
        order_repo = current_domain.repository_for(Order)
        inventory_repo = current_domain.repository_for(Inventory)

        order = order_repo.get(command.order_id)
        if order.status == OrderStatus.CONFIRMED.value:
            # A repeated PlaceOrder would raise OrderConfirmed a second time.
            return

        # Load the stock record of each ordered product. A product with no
        # record is left out, and the service rejects the order.
        inventories = []
        for product_id in order.quantities_by_product():
            found = inventory_repo.query.filter(product_id=product_id).all()
            inventories.extend(found.items)

        # Invoke the domain service. The invariants run here
        service = place_order(order, inventories)
        service()

        # Persist only the primary aggregate
        order_repo.add(order)

Invariants

Just like Aggregates and Entities, Domain Services can also have invariants. These invariants validate the state of the aggregates passed to the service method. Unlike in Aggregates, invariants in Domain Services typically deal with validations that span across multiple aggregates.

pre invariants check the state of the aggregates before the service changes the primary aggregate, while post invariants check the state after the change. When a pre-invariant fails, a ValidationError is raised and the change never happens.

Note

It is a good practice to step back and review the business logic placed in the Domain Service now and then. If an invariant does not use multiple aggregates, it is likely that it belongs within an aggregate and not in the service.

A Full-Blown Example

from datetime import UTC, datetime
from enum import Enum

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

domain = Domain()


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


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

    @invariant.post
    def order_should_contain_items(self):
        if not self.items or len(self.items) == 0:
            raise ValidationError({"_entity": ["Order must contain at least one item"]})

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

    def confirm(self):
        self.status = OrderStatus.CONFIRMED.value
        self.raise_(
            OrderConfirmed(
                order_id=self.id,
                items=[
                    {"product_id": item.product_id, "quantity": item.quantity}
                    for item in self.items
                ],
                confirmed_at=datetime.now(UTC),
            )
        )


@domain.event(part_of=Order)
class OrderConfirmed:
    order_id: Identifier(required=True)
    items = List(content_type=Dict)
    confirmed_at: DateTime(required=True)


@domain.entity(part_of=Order)
class OrderItem:
    product_id: Identifier(required=True)
    quantity: Integer()
    price: Float()


@domain.value_object(part_of="Inventory")
class Warehouse:
    location: String()
    contact: String()


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


@domain.aggregate
class Inventory:
    product_id: Identifier(required=True)
    quantity: Integer()
    warehouse = ValueObject(Warehouse)

    def reserve_stock(self, quantity: int):
        self.quantity -= quantity
        self.raise_(
            StockReserved(
                product_id=self.product_id,
                quantity=quantity,
                reserved_at=datetime.now(UTC),
            )
        )


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

        self.order = order
        self.inventories = inventories

    def __call__(self):
        self.order.confirm()

    @invariant.pre
    def inventory_should_have_sufficient_stock(self):
        for product_id, quantity in self.order.quantities_by_product().items():
            inventory = next(
                (i for i in self.inventories if i.product_id == product_id), None
            )
            if inventory is None or inventory.quantity < quantity:
                raise ValidationError({"_service": ["Product is out of stock"]})

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

    @invariant.post
    def order_should_be_confirmed(self):
        if self.order.status != OrderStatus.CONFIRMED.value:
            raise ValidationError({"_service": ["Order was not confirmed"]})

    @invariant.post
    def confirmation_should_list_every_ordered_item(self):
        confirmations = [e for e in self.order._events if isinstance(e, OrderConfirmed)]
        ordered = self.order.quantities_by_product()
        listed: dict[str, int] = {}
        for line in confirmations[-1].items if confirmations else []:
            listed[line["product_id"]] = (
                listed.get(line["product_id"], 0) + line["quantity"]
            )

        if listed != ordered:
            raise ValidationError(
                {"_service": ["The confirmation does not list every ordered item"]}
            )

When an order is placed, Order status has to be CONFIRMED and the stock record of each product in Inventory has to be reduced.

The service checks the stock and the payment method before anything changes, and then confirms the order. It does not touch Inventory. OrderConfirmed lists each ordered product and quantity, and the inventory reduces its stock when it handles the command issued for that event. The post invariants check that the order is confirmed and that the event lists every ordered item, since an item missing from the event would never be reserved.

A Domain Service fits here because the stock rule needs both aggregates. The Order aggregate cannot see Inventory, and a command handler should not hold business rules.

The stock check reads the inventory as it was loaded. Two orders placed at the same time can both pass it. The inventory's command handler should check the stock again before it reserves.