add event dispatcher and module event handling

Author: pgorecki

src/api/tests/test_catalog.py

@@ -76,6 +76,17 @@ def test_catalog_list_with_two_items(api, api_client):
     assert len(response_data) == 2
 
 
+# @pytest.mark.integration
+# def test_catalog_create_draft(api, api_client):
+#     response = api_client.post("/catalog")
+#     assert False
+
+
+def test_catalog_create_draft_fails_due_to_incomplete_data(api, api_client):
+    response = api_client.post("/catalog")
+    assert response.status_code == 422
+
+
 @pytest.mark.integration
 def test_catalog_delete_draft(api, api_client):
     catalog_module = api.container.catalog_module()

src/config/container.py

@@ -4,6 +4,7 @@
 
 from modules.catalog import CatalogModule
 from modules.iam import IamModule
+from seedwork.application.event_dispatcher import InMemoryEventDispatcher
 
 
 def _default(val):
@@ -56,13 +57,16 @@ class Container(containers.DeclarativeContainer):
 
     config = providers.Configuration()
     engine = providers.Singleton(create_engine_once, config)
+    domain_event_dispatcher = InMemoryEventDispatcher()
 
     catalog_module = providers.Factory(
         CatalogModule,
         engine=engine,
+        domain_event_dispatcher=domain_event_dispatcher,
     )
 
     iam_module = providers.Factory(
         IamModule,
         engine=engine,
+        domain_event_dispatcher=domain_event_dispatcher,
     )

src/modules/catalog/__init__.py

@@ -24,7 +24,6 @@ class CatalogModule(BusinessModule):
         PublishListingCommand,
     )
     supported_queries = (GetAllListings, GetListingDetails, GetListingsOfSeller)
-    supported_events = ()
 
     def configure_unit_of_work(self, uow):
         """Here we have a chance to add extra UOW attributes to be injected into command/query handers"""

src/modules/catalog/application/__init__.py

@@ -0,0 +1,8 @@
+from modules.catalog import CatalogModule
+from modules.catalog.domain.events import ListingPublishedEvent
+
+
+def do_nothing_when_listing_published(
+    event: ListingPublishedEvent, module: CatalogModule
+):
+    print("Listing has been published")

src/modules/catalog/application/event/__init__.py

@@ -1,119 +1,8 @@
-import functools
-from collections.abc import Callable
-from inspect import signature
+from seedwork.application.registry import registry
 
-from pydantic.error_wrappers import ValidationError
+# decorators
+command_handler = registry.command_handler
 
-from seedwork.application.commands import Command
-from seedwork.application.queries import Query
-from seedwork.domain.exceptions import BusinessRuleValidationException
-from seedwork.infrastructure.logging import logger
+query_handler = registry.query_handler
 
-from .command_handlers import CommandResult
-from .query_handlers import QueryResult
-
-
-class Registry:
-    def __init__(self):
-        self.command_handlers = {}
-        self.query_handlers = {}
-
-    def register_command_handler(
-        self, command_class: type[Command], handler: Callable, handler_parameters
-    ):
-        logger.info(f"registering command handler for {command_class} as {handler}")
-        self.command_handlers[command_class] = (handler, handler_parameters)
-
-    def get_command_handler_for(self, command_class) -> Callable:
-        assert (
-            command_class in self.command_handlers
-        ), f"handler for {command_class} not registered"
-        return self.command_handlers[command_class][0]
-
-    def get_command_handler_parameters_for(self, command_class) -> dict:
-        return self.command_handlers[command_class][1].copy()
-
-    def register_query_handler(
-        self, query_class: type[Query], handler: Callable, handler_parameters
-    ):
-        logger.info(f"registering query handler for {query_class} as {handler}")
-        self.query_handlers[query_class] = (handler, handler_parameters)
-
-    def get_query_handler_for(self, query_class) -> dict:
-        assert (
-            query_class in self.query_handlers
-        ), f"handler for {query_class} not registered"
-        return self.query_handlers[query_class][0]
-
-    def get_query_handler_parameters_for(self, query_class) -> Callable:
-        return self.query_handlers[query_class][1].copy()
-
-    def clear(self):
-        self.command_handlers.clear()
-
-
-def find_object_of_class(iterable, cls):
-    for item in iterable:
-        if isinstance(item, cls):
-            return item
-    return None
-
-
-registry = Registry()
-
-
-def command_handler(fn):
-    @functools.wraps(fn)
-    def decorator(*args, **kwargs):
-        try:
-            command = find_object_of_class(args, Command) or find_object_of_class(
-                kwargs.items(), Command
-            )
-            print(
-                "handling command",
-                f"{type(command).__module__}.{type(command).__name__}",
-            )
-            return fn(*args, **kwargs)
-        except ValidationError as e:
-            return CommandResult.failure("Validation error", exception=e)
-        except BusinessRuleValidationException as e:
-            return CommandResult.failure("Business rule validation error", exception=e)
-
-    handler_signature = signature(fn)
-    kwargs_iterator = iter(handler_signature.parameters.items())
-    _, first_param = next(kwargs_iterator)
-    command_class = first_param.annotation
-    assert issubclass(
-        command_class, Command
-    ), "The first parameter must be of type Command"
-    handler_parameters = {}
-    for name, param in kwargs_iterator:
-        handler_parameters[name] = param.annotation
-    registry.register_command_handler(command_class, decorator, handler_parameters)
-    return decorator
-
-
-def query_handler(fn):
-    @functools.wraps(fn)
-    def decorator(*args, **kwargs):
-        try:
-            query = find_object_of_class(args, Query) or find_object_of_class(
-                kwargs.items(), Query
-            )
-            print("handling query", f"{type(query).__module__}.{type(query).__name__}")
-            return fn(*args, **kwargs)
-        except ValidationError as e:
-            return QueryResult.failed("Validation error", exception=e)
-        except BusinessRuleValidationException as e:
-            return QueryResult.failed("Business rule validation error", exception=e)
-
-    handler_signature = signature(fn)
-    kwargs_iterator = iter(handler_signature.parameters.items())
-    _, first_param = next(kwargs_iterator)
-    query_class = first_param.annotation
-    assert issubclass(query_class, Query), "The first parameter must be of type Command"
-    handler_parameters = {}
-    for name, param in kwargs_iterator:
-        handler_parameters[name] = param.annotation
-    registry.register_query_handler(query_class, decorator, handler_parameters)
-    return decorator
+domain_event_handler = registry.domain_event_handler

src/modules/catalog/application/event/do_nothing_when_listing_published.py

@@ -0,0 +1,29 @@
+import abc
+from collections import defaultdict
+
+from seedwork.domain.events import Event
+
+
+class EventDispatcher(metaclass=abc.ABCMeta):
+    """An interface for a generic event dispatcher"""
+
+    @abc.abstractmethod
+    def add_event_handler(self, event_class: type[Event], event_handler: callable):
+        raise NotImplementedError()
+
+    @abc.abstractmethod
+    def dispatch(self):
+        raise NotImplementedError()
+
+
+class InMemoryEventDispatcher(EventDispatcher):
+    def __init__(self):
+        self._handlers = defaultdict(set)
+
+    def add_event_handler(self, event_class: type[Event], event_handler: callable):
+        self._handlers[event_class].add(event_handler)
+
+    def dispatch(self, event: type[Event]):
+        event_class = type(event)
+        for event_handler in self._handlers[event_class]:
+            event_handler(event)

src/seedwork/application/decorators.py

@@ -1,3 +1,4 @@
+from seedwork.application.event_dispatcher import EventDispatcher
 from seedwork.infrastructure.logging import logger
 
 
@@ -26,7 +27,8 @@ def handle(module, query_or_command, *args, **kwargs):
 
 from sqlalchemy.orm import Session
 
-from seedwork.application.decorators import registry
+from seedwork.application.decorators import registry as default_registry
+from seedwork.domain.events import DomainEvent
 from seedwork.domain.repositories import GenericRepository
 from seedwork.infrastructure.request_context import request_context
 
@@ -61,11 +63,34 @@ class BusinessModule:
     unit_of_work_class = UnitOfWork
     supported_commands = ()
     supported_queries = ()
-    supported_events = ()
+    event_handlers = ()
+    registry = default_registry
 
-    def __init__(self, **kwargs):
+    def __init__(self, domain_event_dispatcher: type[EventDispatcher], **kwargs):
         self._uow: ContextVar[UnitOfWork] = ContextVar("_uow", default=None)
         self.init_kwargs = kwargs
+        self._domain_event_dispatcher = domain_event_dispatcher
+        self.register_event_handlers()
+
+    def register_event_handlers(self):
+        """Registers all event handlers declared in this module"""
+        if self._domain_event_dispatcher is None:
+            return
+
+        for event_class in self.get_handleable_domain_events():
+            self._domain_event_dispatcher.add_event_handler(
+                event_class=event_class, event_handler=self.handle_domain_event
+            )
+
+    def get_handleable_domain_events(self) -> list[type[DomainEvent]]:
+        """Returns a list of domain event classes that this module is capable of handling"""
+        handled_event_types = set()
+        for handler in self.event_handlers:
+            event_class, handler_parameters = self.registry.inspect_handler_parameters(
+                handler
+            )
+            handled_event_types.add(event_class)
+        return handled_event_types
 
     @contextmanager
     def unit_of_work(self, **kwargs):
@@ -84,7 +109,7 @@ def unit_of_work(self, **kwargs):
             request_context.correlation_id.set(None)
 
     def create_unit_of_work(self, correlation_id, db_session):
-        """Unit of Work factory"""
+        """Unit of Work factory, creates new unit of work"""
         uow = self.unit_of_work_class(
             module=self,
             correlation_id=correlation_id,
@@ -94,11 +119,11 @@ def create_unit_of_work(self, correlation_id, db_session):
         return uow
 
     def get_unit_of_work_init_kwargs(self):
-        """Provide additional kwargs for Unit of Work if you are using a custom one"""
+        """Returns additional kwargs used for initialization of new Unit of Work"""
         return dict()
 
     def configure_unit_of_work(self, uow):
-        """Allows to alter Unit of Work (i.e. add extra attributes)"""
+        """Allows to alter Unit of Work (i.e. add extra attributes) after it is instantiated"""
 
     def end_unit_of_work(self, uow):
         uow.db_session.commit()
@@ -107,35 +132,54 @@ def configure(self, **kwargs):
         self.init_kwargs = kwargs
 
     def execute_command(self, command):
+        """Module entrypoint. Use it to change the state of the module by passing a command object"""
         command_class = type(command)
         assert (
             command_class in self.supported_commands
         ), f"{command_class} is not included in {type(self).__name__}.supported_commands"
-        handler = registry.get_command_handler_for(command_class)
-        kwarg_params = registry.get_command_handler_parameters_for(command_class)
+        handler = self.registry.get_command_handler_for(command_class)
+        kwarg_params = self.registry.get_command_handler_parameters_for(command_class)
         kwargs = self.resolve_handler_kwargs(kwarg_params)
-        return handler(command, **kwargs)
+        command_result = handler(command, **kwargs)
+        if command_result.is_success():
+            self.publish_domain_events(command_result.events)
+        return command_result
 
     def execute_query(self, query):
+        """Module entrypoint. Use it to read the state of the module by passing a query object"""
         query_class = type(query)
         assert (
             query_class in self.supported_queries
         ), f"{query_class} is not included in {type(self).__name__}.supported_queries"
-        handler = registry.get_query_handler_for(query_class)
-        kwarg_params = registry.get_query_handler_parameters_for(query_class)
+        handler = self.registry.get_query_handler_for(query_class)
+        kwarg_params = self.registry.get_query_handler_parameters_for(query_class)
         kwargs = self.resolve_handler_kwargs(kwarg_params)
         return handler(query, **kwargs)
 
     @property
     def uow(self) -> UnitOfWork:
+        """Get current unit of work. Use self.unit_of_work() to create a new instance of UoW"""
         uow = self._uow.get()
         assert uow, "Unit of work not set, use context manager"
         return uow
 
     def resolve_handler_kwargs(self, kwarg_params) -> dict:
+        """Match kwargs required by a function to attributes available in a unit of work"""
         kwargs = {}
         for param_name, param_type in kwarg_params.items():
             for attr in self.uow.__dict__.values():
                 if isinstance(attr, param_type):
                     kwargs[param_name] = attr
         return kwargs
+
+    def publish_domain_events(self, events):
+        ...
+
+    def handle_domain_event(self, event: type[DomainEvent]):
+        """Execute all registered handlers within this module for this event type"""
+        for handler in self.event_handlers:
+            event_class, handler_parameters = self.registry.inspect_handler_parameters(
+                handler
+            )
+            if event_class is type(event):
+                handler(event, self)