Application Services

Application services act as a bridge between the external API layer and the domain model, orchestrating 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.

Key Facts

• Application Services encapsulate business use cases and serve as the main entry point for external requests to interact with the domain model.
• Application Services are predominantly used on the write side of the application. If you want to use them on the read side as well, it is recommended to create a separate application service for the read side.
• Application Services are stateless and should not hold any business logic themselves; instead, they orchestrate and manage the flow of data and operations to and from the domain model.
• Application Services ensure transaction consistency by automatically enclosing all use case methods within a unit of work context.
• Application Services can interact with multiple aggregates and repositories, but should only persist one aggregate, relying on events for eventual consistency.

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(__file__, "Auth", load_toml=False)


@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(sefl, 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)