Application Service Enhancements

src/protean/__init__.py

@@ -1,6 +1,7 @@
 __version__ = "0.12.1"
 
 from .core.aggregate import apply, atomic_change
+from .core.application_service import use_case
 from .core.entity import invariant
 from .core.event import BaseEvent
 from .core.model import BaseModel
@@ -13,19 +14,20 @@
 from .utils.mixins import handle
 
 __all__ = [
+    "apply",
+    "atomic_change",
     "BaseEvent",
     "BaseModel",
+    "current_domain",
+    "current_uow",
     "Domain",
     "Engine",
-    "Q",
-    "QuerySet",
-    "UnitOfWork",
-    "apply",
+    "g",
     "get_version",
     "handle",
     "invariant",
-    "atomic_change",
-    "current_domain",
-    "current_uow",
-    "g",
+    "Q",
+    "QuerySet",
+    "UnitOfWork",
+    "use_case",
 ]

src/protean/core/application_service.py

@@ -1,6 +1,8 @@
+import functools
 import logging
 
-from protean.exceptions import NotSupportedError
+from protean.core.unit_of_work import UnitOfWork
+from protean.exceptions import IncorrectUsageError, NotSupportedError
 from protean.utils import DomainObjects, derive_element_class
 from protean.utils.container import Element, OptionsMixin
 
@@ -28,8 +30,37 @@ def __new__(cls, *args, **kwargs):
 
     @classmethod
     def _default_options(cls):
-        return []
+        return [("part_of", None)]
 
 
 def application_service_factory(element_cls, domain, **opts):
-    return derive_element_class(element_cls, BaseApplicationService, **opts)
+    element_cls = derive_element_class(element_cls, BaseApplicationService, **opts)
+
+    if not element_cls.meta_.part_of:
+        raise IncorrectUsageError(
+            f"Application Service `{element_cls.__name__}` needs to be associated with an aggregate"
+        )
+
+    return element_cls
+
+
+def use_case(func):
+    """Decorator to mark a method as a use case in an Application Service.
+
+    Args:
+        func (Callable): The method to be decorated.
+
+    Returns:
+        Callable: 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

src/protean/core/value_object.py

@@ -28,37 +28,37 @@ def _default_options(cls):
             ("part_of", None),
         ]
 
-    def __init_subclass__(subclass) -> None:
+    def __init_subclass__(cls) -> None:
         super().__init_subclass__()
 
         # Record invariant methods
-        setattr(subclass, "_invariants", defaultdict(dict))
+        setattr(cls, "_invariants", defaultdict(dict))
 
-        subclass.__validate_for_basic_field_types()
-        subclass.__validate_for_non_identifier_fields()
-        subclass.__validate_for_non_unique_fields()
+        cls.__validate_for_basic_field_types()
+        cls.__validate_for_non_identifier_fields()
+        cls.__validate_for_non_unique_fields()
 
     @classmethod
-    def __validate_for_basic_field_types(subclass):
-        for field_name, field_obj in fields(subclass).items():
+    def __validate_for_basic_field_types(cls):
+        for field_name, field_obj in fields(cls).items():
             # Value objects can hold all kinds of fields, except associations
             if isinstance(field_obj, (Reference, Association)):
                 raise IncorrectUsageError(
                     f"Value Objects cannot have associations. "
-                    f"Remove {field_name} ({field_obj.__class__.__name__}) from class {subclass.__name__}"
+                    f"Remove {field_name} ({field_obj.__class__.__name__}) from class {cls.__name__}"
                 )
 
     @classmethod
-    def __validate_for_non_identifier_fields(subclass):
-        for field_name, field_obj in fields(subclass).items():
+    def __validate_for_non_identifier_fields(cls):
+        for field_name, field_obj in fields(cls).items():
             if field_obj.identifier:
                 raise IncorrectUsageError(
                     f"Value Objects cannot contain fields marked 'identifier' (field '{field_name}')"
                 )
 
     @classmethod
-    def __validate_for_non_unique_fields(subclass):
-        for field_name, field_obj in fields(subclass).items():
+    def __validate_for_non_unique_fields(cls):
+        for field_name, field_obj in fields(cls).items():
             if field_obj.unique:
                 raise IncorrectUsageError(
                     f"Value Objects cannot contain fields marked 'unique' (field '{field_name}')"

src/protean/domain/__init__.py

@@ -489,6 +489,7 @@ def _register_element(self, element_type, element_cls, **opts):  # noqa: C901
 
         # 2. Meta Linkages
         if element_type in [
+            DomainObjects.APPLICATION_SERVICE,
             DomainObjects.ENTITY,
             DomainObjects.EVENT,
             DomainObjects.EVENT_HANDLER,

tests/application_service/test_application_service_call.py

@@ -0,0 +1,54 @@
+from protean.core.aggregate import BaseAggregate
+from protean.core.application_service import BaseApplicationService, use_case
+from protean.core.event import BaseEvent
+from protean.fields import Identifier, String
+from protean.utils.globals import current_domain
+
+
+class User(BaseAggregate):
+    email = String()
+    name = String()
+    status = String(choices=["INACTIVE", "ACTIVE", "ARCHIVED"], default="INACTIVE")
+
+    def activate(self):
+        self.status = "ACTIVE"
+
+
+class Registered(BaseEvent):
+    user_id = Identifier()
+    email = String()
+    name = String()
+
+
+class UserApplicationServices(BaseApplicationService):
+    @use_case
+    def register_user(self, email: str, name: str) -> Identifier:
+        user = User(email=email, name=name)
+        user.raise_(Registered(user_id=user.id, email=user.email, name=user.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)
+
+
+def test_application_service_method_invocation(test_domain):
+    test_domain.register(User)
+    test_domain.register(UserApplicationServices, part_of=User)
+    test_domain.register(Registered, part_of=User)
+    test_domain.init(traverse=False)
+
+    app_services_obj = UserApplicationServices()
+
+    user_id = app_services_obj.register_user(
+        email="[email protected]", name="John Doe"
+    )
+    assert user_id is not None
+
+    app_services_obj.activate_user(user_id)
+    user = current_domain.repository_for(User).get(user_id)
+    assert user.status == "ACTIVE"

tests/application_service/test_application_service_options.py

@@ -0,0 +1,50 @@
+import pytest
+
+from protean.core.aggregate import BaseAggregate
+from protean.core.application_service import BaseApplicationService
+from protean.core.event import BaseEvent
+from protean.core.event_handler import BaseEventHandler
+from protean.exceptions import IncorrectUsageError, NotSupportedError
+from protean.fields import Identifier, String
+from protean.utils.globals import current_domain
+
+
+class User(BaseAggregate):
+    email = String()
+    name = String()
+
+
+def test_that_base_command_handler_cannot_be_instantianted():
+    with pytest.raises(NotSupportedError):
+        BaseApplicationService()
+
+
+def test_part_of_specified_during_registration(test_domain):
+    class UserApplicationService(BaseApplicationService):
+        pass
+
+    test_domain.register(UserApplicationService, part_of=User)
+    assert UserApplicationService.meta_.part_of == User
+
+
+def test_part_of_defined_via_annotation(
+    test_domain,
+):
+    @test_domain.application_service(part_of=User)
+    class UserApplicationService:
+        pass
+
+    assert UserApplicationService.meta_.part_of == User
+
+
+def test_part_of_is_mandatory(test_domain):
+    class UserApplicationService(BaseApplicationService):
+        pass
+
+    with pytest.raises(IncorrectUsageError) as exc:
+        test_domain.register(UserApplicationService)
+
+    assert (
+        exc.value.args[0]
+        == "Application Service `UserApplicationService` needs to be associated with an aggregate"
+    )

tests/application_service/tests.py → tests/application_service/test_initialization.py

@@ -1,10 +1,18 @@
 import pytest
 
+from protean.core.aggregate import BaseAggregate
 from protean.core.application_service import BaseApplicationService
 from protean.exceptions import NotSupportedError
 from protean.utils import fully_qualified_name
 
-from .elements import DummyApplicationService
+
+class DummyAggregate(BaseAggregate):
+    pass
+
+
+class DummyApplicationService(BaseApplicationService):
+    def do_application_process(self):
+        print("Performing application process...")
 
 
 class TestApplicationServiceInitialization:
@@ -19,7 +27,7 @@ def test_that_application_service_can_be_instantiated(self):
 
 class TestApplicationServiceRegistration:
     def test_that_application_service_can_be_registered_with_domain(self, test_domain):
-        test_domain.register(DummyApplicationService)
+        test_domain.register(DummyApplicationService, part_of=DummyAggregate)
 
         assert (
             fully_qualified_name(DummyApplicationService)
@@ -29,7 +37,7 @@ def test_that_application_service_can_be_registered_with_domain(self, test_domai
     def test_that_application_service_can_be_registered_via_annotations(
         self, test_domain
     ):
-        @test_domain.application_service
+        @test_domain.application_service(part_of=DummyAggregate)
         class AnnotatedApplicationService:
             def special_method(self):
                 pass
@@ -38,3 +46,12 @@ def special_method(self):
             fully_qualified_name(AnnotatedApplicationService)
             in test_domain.registry.application_services
         )
+
+    def test_that_application_service_part_of_is_resolve_on_domain_init(
+        self, test_domain
+    ):
+        test_domain.register(DummyAggregate)
+        test_domain.register(DummyApplicationService, part_of="DummyAggregate")
+        test_domain.init(traverse=False)
+
+        assert DummyApplicationService.meta_.part_of == DummyAggregate

tests/application_service/test_uow_around_application_services.py

@@ -0,0 +1,52 @@
+import mock
+
+from protean.core.aggregate import BaseAggregate
+from protean.core.application_service import BaseApplicationService, use_case
+from protean.core.event import BaseEvent
+from protean.fields import Identifier, String
+from protean.utils.globals import current_domain
+
+
+class User(BaseAggregate):
+    email = String()
+    name = String()
+
+
+class Registered(BaseEvent):
+    user_id = Identifier()
+    email = String()
+    name = String()
+
+
+class UserApplicationServices(BaseApplicationService):
+    @use_case
+    def register_user(self, email: str, name: str) -> Identifier:
+        user = User(email=email, name=name)
+        user.raise_(Registered(user_id=user.id, email=user.email, name=user.name))
+        current_domain.repository_for(User).add(user)
+
+        return user.id
+
+
+@mock.patch("protean.utils.mixins.UnitOfWork.__enter__")
+@mock.patch("protean.utils.mixins.UnitOfWork.__exit__")
+def test_that_method_is_enclosed_in_uow(mock_exit, mock_enter, test_domain):
+    test_domain.register(User)
+    test_domain.register(UserApplicationServices, part_of=User)
+    test_domain.register(Registered, part_of=User)
+    test_domain.init(traverse=False)
+
+    mock_parent = mock.Mock()
+
+    mock_parent.attach_mock(mock_enter, "m1")
+    mock_parent.attach_mock(mock_exit, "m2")
+
+    app_services_obj = UserApplicationServices()
+    app_services_obj.register_user(email="[email protected]", name="John Doe")
+
+    mock_parent.assert_has_calls(
+        [
+            mock.call.m1(),
+            mock.call.m2(None, None, None),
+        ]
+    )