Services

Application services orchestrate use cases for external callers. Domain services encapsulate business rules that span multiple aggregates.

Guides: Application Services · Domain Services

---

BaseApplicationService

Bases: Element, OptionsMixin

Base class for application services -- stateless orchestration layers that coordinate use cases between external callers (API controllers, CLI handlers, background jobs) and the domain model.

Application services load aggregates, invoke domain methods, and persist results without containing business logic themselves. They are always associated with one aggregate via part_of. Use the @use_case decorator on methods for automatic UnitOfWork wrapping.

Unlike command handlers, application services are invoked directly (not via domain.process()) and always return values synchronously.

Meta Options

Option	Type	Description
part_of	type	The aggregate class this service orchestrates. Required.

Example::

@domain.application_service(part_of=Order)
class OrderService(BaseApplicationService):

    @use_case
    def place_order(self, order_data: dict) -> Order:
        order = Order(**order_data)
        repo = domain.repository_for(Order)
        repo.add(order)
        return order

use_case

use_case(func: Callable[..., Any]) -> Callable[..., Any]

Decorator to mark a method as a use case in an Application Service.

PARAMETER	DESCRIPTION
func	The method to be decorated. TYPE: Callable[..., Any]

RETURNS	DESCRIPTION
Callable[..., Any]	The decorated method.

Source code in src/protean/core/application_service.py

def use_case(func: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator to mark a method as a use case in an Application Service.

    Args:
        func: The method to be decorated.

    Returns:
        The decorated method.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        logger.info(f"Executing use case: {func.__name__}")

        # Wrap in a Unit of Work context
        with UnitOfWork():
            return func(*args, **kwargs)

    setattr(wrapper, "_use_case", True)  # Mark the method as a use case
    return wrapper

---

BaseDomainService

BaseDomainService(*aggregates: Union[BaseAggregate, List[BaseAggregate]])

Bases: Element, OptionsMixin

Base class for domain services that encapsulate business logic spanning multiple aggregates.

Domain services are stateless, instantiated with the aggregate instances they operate on, and must be associated with two or more aggregates via the part_of option. Public methods (including __call__) are automatically wrapped with pre/post invariant checks when the service class has methods decorated with @invariant.pre or @invariant.post.

Meta Options

Option	Type	Description
part_of	list	List of two or more aggregate classes this service operates on. Required.

Example::

@domain.domain_service(part_of=[Order, Inventory])
class PlaceOrderService(BaseDomainService):

    @invariant.post
    def order_should_have_items(self):
        if not self._aggregates[0].items:
            raise ValidationError({"items": ["Order must have items"]})

    def __call__(self):
        order, inventory = self._aggregates
        inventory.reserve(order.items)
        order.confirm()

Initialize a DomainService with one or more aggregates.

PARAMETER	DESCRIPTION
*aggregates	One or more aggregates to operate on. TYPE: Union[BaseAggregate, List[BaseAggregate]] DEFAULT: ()

Source code in src/protean/core/domain_service.py

def __init__(self, *aggregates: Union[BaseAggregate, List[BaseAggregate]]):
    """Initialize a DomainService with one or more aggregates.

    Args:
        *aggregates: One or more aggregates to operate on.
    """
    self._aggregates = aggregates