Application Services

DDD

Application services connect the external API layer to the domain model. They orchestrate business logic and use cases without exposing the underlying domain complexity. They encapsulate and coordinate operations, making them reusable and easier to manage, ensuring that all interactions with the domain are consistent and controlled.

For background on how application services orchestrate use cases, see Application Services concept.

Defining an Application Service

Application Services are defined with the Domain.application_service decorator:

from protean import Domain, current_domain, use_case
from protean.fields import Identifier, String

auth = Domain(name="Auth")


@auth.aggregate
class User:
    email: String()
    name: String()
    status: String(choices=["INACTIVE", "ACTIVE", "ARCHIVED"], default="INACTIVE")

    @classmethod
    def register(cls, email: str, name: str):
        user = cls(email=email, name=name)
        user.raise_(Registered(user_id=user.id, email=user.email, name=user.name))

        return user

    def activate(self):
        self.status = "ACTIVE"


@auth.event(part_of=User)
class Registered:
    user_id: Identifier()
    email: String()
    name: String()


@auth.application_service(part_of=User)
class UserApplicationServices:
    @use_case
    def register_user(self, email: str, name: str) -> Identifier:
        user = User.register(email, name)
        current_domain.repository_for(User).add(user)

        return user.id

    @use_case
    def activate_user(self, user_id: Identifier) -> None:
        user = current_domain.repository_for(User).get(user_id)
        user.activate()
        current_domain.repository_for(User).add(user)


auth.register(User)
auth.register(UserApplicationServices, part_of=User)
auth.register(Registered, part_of=User)
auth.init(traverse=False)

The @auth.application_service(part_of=User) decorator registers the class as an Application Service associated with the User aggregate. The part_of option is mandatory. Every Application Service must be linked to exactly one aggregate.
The @use_case decorator on register_user marks the method as a use case entry point. It automatically wraps the method body in a UnitOfWork context, providing transaction management.
The @use_case decorator on activate_user follows the same pattern. Each use case method represents a distinct business operation, receives plain Python arguments, and can return values directly to the caller.

Unlike command handlers (which are dispatched via domain.process()), application services are instantiated and called directly. A use case runs inside a Unit of Work, so call it after the domain is initialized (the example above ends with auth.init(traverse=False)). Call it inside a domain context too, such as with auth.domain_context():. Without one, the call raises ConfigurationError:

def register_and_activate():
    user_service = UserApplicationServices()
    user_id = user_service.register_user(email="john@example.com", name="John Doe")
    user_service.activate_user(user_id=user_id)
    return user_id

Because you call application services directly, they are ideal for synchronous request-response workflows where the caller needs an immediate result.

The `@use_case` Decorator

The @use_case decorator is imported from protean and is the primary mechanism for marking methods as use case entry points:

from protean import use_case

It does the following:

Unit of Work wrapping: Every method decorated with @use_case is automatically enclosed in a UnitOfWork context. If the method completes successfully, the UoW commits all pending changes. If an exception is raised, the UoW rolls back all changes.
Execution logging: The decorator logs the invocation of each use case at the INFO level, providing traceability for debugging and auditing.
Domain context check: The decorator raises ConfigurationError when no domain context is active, because the Unit of Work needs a domain to open against.

@domain.aggregate
class Order:
    customer_id: Identifier(required=True)
    items: List(content_type=String)

    @classmethod
    def create(cls, customer_id: str, items: list) -> "Order":
        return cls(customer_id=customer_id, items=items)


@domain.application_service(part_of=Order)
class OrderApplicationServices:
    @use_case
    def place_order(self, customer_id: str, items: list) -> Identifier:
        # Everything inside here runs within a UnitOfWork
        order = Order.create(customer_id=customer_id, items=items)
        current_domain.repository_for(Order).add(order)
        return order.id

Note

Only methods decorated with @use_case are automatically wrapped in a Unit of Work. Regular methods on an Application Service class do not receive this treatment. You can add helper methods without the decorator, but they will not have automatic transaction management.

Internal workflow

Sequence diagram

API Layer Calls Use Case Method: The external layer (e.g., a REST controller, GraphQL resolver, or CLI handler) instantiates the Application Service and calls the appropriate use case method with plain Python arguments.
UnitOfWork Begins: The @use_case decorator automatically opens a UnitOfWork context before the method body executes.
Application Service Loads Aggregate: If the use case operates on an existing aggregate, the service retrieves it from the repository using current_domain.repository_for(Aggregate).get(id).
Aggregate Receives Method Call: The application service invokes the relevant domain method on the aggregate, passing the necessary parameters.
Aggregate Mutates: The aggregate applies business rules, validates invariants, and changes its internal state. It may also raise domain events.
Application Service Persists Aggregate: The service persists the mutated aggregate back to the repository using repository.add().
UnitOfWork Commits: Upon successful completion of the method, the UnitOfWork commits all pending changes to the persistence store and publishes any raised domain events.
Return Value Delivered: The return value from the use case method is passed directly back to the calling API layer, enabling immediate feedback (e.g., a newly created entity ID).

Return Values

Application Services execute synchronously and always return values directly to the caller. This is one of their key advantages for request-response workflows.

Common return value patterns:

@domain.application_service(part_of=User)
class UserApplicationServices:
    @use_case
    def register_user(self, email: str, name: str) -> Identifier:
        user = User.register(email, name)
        current_domain.repository_for(User).add(user)
        return user.id  # Return the new entity's identifier

    @use_case
    def activate_user(self, user_id: Identifier) -> None:
        user = current_domain.repository_for(User).get(user_id)
        user.activate()
        current_domain.repository_for(User).add(user)
        # No return value needed for mutative operations

    @use_case
    def get_user(self, user_id: Identifier) -> User:
        return current_domain.repository_for(User).get(user_id)

This direct return capability distinguishes Application Services from both Command Handlers (whose return values depend on sync/async processing mode) and Event Handlers (which never return values). Application Services always execute synchronously, so the caller always receives the return value immediately.

Unit of Work

Use case methods always execute within a UnitOfWork context, if the method completes successfully, all changes are committed; if an exception is raised, everything is rolled back.

For details on how the Unit of Work pattern works, see the Unit of Work guide.

Note

A UnitOfWork context applies to objects in the aggregate cluster, and not multiple aggregates. An Application Service method can load multiple aggregates to perform the business process, but should never persist more than one at a time. Event handlers should sync the other aggregates eventually, in response to domain events.

Error Handling

Application Services handle errors differently from Command and Event Handlers. Because they execute synchronously with the caller always present, there is no handle_error hook method. Instead, exceptions propagate directly to the caller.

When an exception occurs inside a @use_case method, the UnitOfWork context manager rolls back all uncommitted changes before re-raising the exception. This ensures that no partial state is persisted, maintaining data consistency.

from protean.exceptions import ValidationError


@domain.application_service(part_of=User)
class UserApplicationServices:
    @use_case
    def register_user(self, email: str, name: str) -> Identifier:
        user = User.register(email, name)
        current_domain.repository_for(User).add(user)
        return user.id


# In the API layer
def register_john():
    try:
        user_service = UserApplicationServices()
        user_id = user_service.register_user(email="john@example.com", name="John Doe")
        return 201, user_id
    except ValidationError as exc:
        # The input broke a rule, such as a taken email
        return 400, exc.messages
    except Exception:
        # Anything else is unexpected
        return 500, "Internal server error"

Best Practices:

Keep use case methods focused on orchestration so that domain-level errors are raised by the aggregate, not the service.
Let domain validation errors propagate to the API layer, where you can map them to the appropriate HTTP status.
Use the API layer for cross-cutting concerns like logging failed operations, not the application service itself.

Testing Application Services

Application services are tested by calling use case methods directly and asserting on the return values and persisted state:

# fragment
def test_register_user(test_domain):
    service = UserApplicationServices()
    user_id = service.register_user(email="jane@example.com", name="Jane Doe")

    assert user_id is not None
    user = test_domain.repository_for(User).get(user_id)
    assert user.email == "jane@example.com"
    assert user.status == "INACTIVE"

Because application services always execute synchronously, no special configuration is needed. You call the method and assert on the result.