domain event handling across modules

Author: pgorecki

src/seedwork/application/command_handlers.py

@@ -28,8 +28,14 @@ def failure(cls, message="Failure", exception=None) -> "CommandResult":
         return result
 
     @classmethod
-    def success(cls, entity_id=None, payload=None, events=[]) -> "CommandResult":
+    def success(
+        cls, entity_id=None, payload=None, event=None, events=None
+    ) -> "CommandResult":
         """Creates a successful result"""
+        if events is None:
+            events = []
+        if event:
+            events.append(event)
         return cls(entity_id=entity_id, payload=payload, events=events)
 
 

src/seedwork/application/event_dispatcher.py

@@ -23,7 +23,7 @@ def __init__(self):
     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]):
+    def dispatch(self, event: type[Event], sender: any = None):
         event_class = type(event)
         for event_handler in self._handlers[event_class]:
-            event_handler(event)
+            event_handler(event, sender)

src/seedwork/application/modules.py

@@ -98,7 +98,7 @@ def unit_of_work(self, **kwargs):
         engine = get_arg("engine", kwargs, self.init_kwargs)
         correlation_id = uuid.uuid4()
         with Session(engine) as db_session:
-            uow = self.create_unit_of_work(correlation_id, db_session)
+            uow = self.create_unit_of_work(correlation_id, db_session, kwargs)
             self.configure_unit_of_work(uow)
             request_context.correlation_id.set(correlation_id)
             self._uow.set(uow)
@@ -108,29 +108,27 @@ def unit_of_work(self, **kwargs):
             self._uow.set(None)
             request_context.correlation_id.set(None)
 
-    def create_unit_of_work(self, correlation_id, db_session):
+    def create_unit_of_work(self, correlation_id, db_session, kwargs):
         """Unit of Work factory, creates new unit of work"""
         uow = self.unit_of_work_class(
             module=self,
             correlation_id=correlation_id,
             db_session=db_session,
             **self.get_unit_of_work_init_kwargs(),
+            **kwargs,
         )
         return uow
 
     def get_unit_of_work_init_kwargs(self):
         """Returns additional kwargs used for initialization of new Unit of Work"""
-        return dict()
+        return self.init_kwargs
 
     def configure_unit_of_work(self, uow):
         """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()
 
-    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)
@@ -160,26 +158,43 @@ def execute_query(self, query):
     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"
+        assert uow, f"Unit of work not set in {self}, 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
+            for attr_name, attr_value in self.uow.__dict__.items():
+                if attr_name == param_name or isinstance(attr_value, param_type):
+                    kwargs[param_name] = attr_value
         return kwargs
 
     def publish_domain_events(self, events):
-        ...
+        for event in events:
+            self._domain_event_dispatcher.dispatch(event=event, sender=self)
 
-    def handle_domain_event(self, event: type[DomainEvent]):
+    def handle_domain_event(
+        self, event: type[DomainEvent], sender: type["BusinessModule"]
+    ):
         """Execute all registered handlers within this module for this event type"""
+        event_was_sent_by_other_module = self is not sender
+
+        if event_was_sent_by_other_module:
+            # The sender executed the command that resulted in an event being published.
+            # as a rule of thumb we want to handle the domain event within same transaction,
+            # thus within same Unit of Work.
+            # Therefore, the receiver must use UoW of sender
+            original_uow = self._uow.get()
+            self._uow.set(sender.uow)
+
         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)
+
+        if event_was_sent_by_other_module:
+            # restore the original UoW
+            self._uow.set(original_uow)

src/seedwork/application/registry.py

@@ -131,7 +131,11 @@ def domain_event_handler(self, fn: Callable):
 
         @functools.wraps(fn)
         def decorator(*args, **kwargs):
-            raise not NotImplementedError()
+            event = find_object_of_class(args, DomainEvent) or find_object_of_class(
+                kwargs.items(), DomainEvent
+            )
+            print("handling event", f"{type(event).__module__}.{type(event).__name__}")
+            return fn(*args, **kwargs)
 
         domain_event_class, handler_parameters = self.inspect_handler_parameters(fn)
         assert issubclass(

src/seedwork/tests/application/test_mono_module_branching.py

@@ -0,0 +1,152 @@
+from dataclasses import dataclass
+
+import pytest
+
+from seedwork.application.command_handlers import CommandResult
+from seedwork.application.commands import Command
+from seedwork.application.event_dispatcher import InMemoryEventDispatcher
+from seedwork.application.modules import BusinessModule, UnitOfWork
+from seedwork.application.registry import Registry
+from seedwork.domain.events import DomainEvent
+
+registry = Registry()
+
+
+@dataclass
+class CompleteOrderCommand(Command):
+    order_id: str
+
+
+@dataclass
+class ProcessPaymentCommand(Command):
+    order_id: str
+
+
+@dataclass
+class ShipOrderCommand(Command):
+    order_id: str
+
+
+class OrderCompletedEvent(DomainEvent):
+    order_id: str
+
+
+class PaymentProcessedEvent(DomainEvent):
+    order_id: str
+
+
+class OrderShippedEvent(DomainEvent):
+    order_id: str
+
+
+@registry.command_handler
+def complete_order(command: CompleteOrderCommand, history):
+    history.append(f"completing {command.order_id}")
+    return CommandResult.success(
+        payload=None, event=OrderCompletedEvent(order_id=command.order_id)
+    )
+
+
+@registry.command_handler
+def process_payment(command: ProcessPaymentCommand, history):
+    history.append(f"processing payment for {command.order_id}")
+    return CommandResult.success(
+        payload=None, event=PaymentProcessedEvent(order_id=command.order_id)
+    )
+
+
+@registry.command_handler
+def ship_order(command: ShipOrderCommand, history):
+    history.append(f"shipping {command.order_id}")
+    return CommandResult.success(
+        payload=None, event=OrderShippedEvent(order_id=command.order_id)
+    )
+
+
+@registry.domain_event_handler
+def when_order_is_completed_process_payment_policy(
+    event: OrderCompletedEvent, module: type[BusinessModule]
+):
+    module.uow.history.append(
+        f"starting when_order_is_completed_process_payment_policy for {event.order_id}"
+    )
+    module.execute_command(ProcessPaymentCommand(order_id=event.order_id))
+
+
+@registry.domain_event_handler
+def when_order_is_completed_ship_order_policy(
+    event: OrderCompletedEvent, module: type[BusinessModule]
+):
+    module.uow.history.append(
+        f"starting when_order_is_completed_ship_order_policy for {event.order_id}"
+    )
+    module.execute_command(ShipOrderCommand(order_id=event.order_id))
+
+
+@registry.domain_event_handler
+def when_payment_is_processed_open_champagne_policy(
+    event: PaymentProcessedEvent, module: type[BusinessModule]
+):
+    module.uow.history.append(
+        f"starting when_payment_is_processed_open_champagne_policy for {event.order_id}"
+    )
+
+
+@registry.domain_event_handler
+def when_order_is_shipped_sit_and_relax_policy(
+    event: OrderShippedEvent, module: type[BusinessModule]
+):
+    module.uow.history.append(
+        f"starting when_order_is_shipped_sit_and_relax_policy for {event.order_id}"
+    )
+
+
+@dataclass
+class MonoUnitOfWork(UnitOfWork):
+    history: list
+
+
+class MonoModule(BusinessModule):
+    registry = registry
+    unit_of_work_class = MonoUnitOfWork
+    supported_commands = (CompleteOrderCommand, ProcessPaymentCommand, ShipOrderCommand)
+    supported_queries = ()
+    event_handlers = (
+        when_order_is_completed_process_payment_policy,
+        when_order_is_completed_ship_order_policy,
+        when_payment_is_processed_open_champagne_policy,
+        when_order_is_shipped_sit_and_relax_policy,
+    )
+
+
+@pytest.mark.integration
+def test_mono_module_command_branching_flow():
+    """This tests the branching code flow:
+                                    CompleteOrderCommand
+                                            ↓
+                                    OrderCompletedEvent
+                                    ↓                 ↓
+    when_order_is_completed_process_payment_policy    when_order_is_completed_ship_order_policy
+             ↓                                              ↓
+    ProcessPaymentCommand                             ShipOrderCommand
+             ↓                                              ↓
+    PaymentProcessedEvent                             OrderShippedEvent
+             ↓                                              ↓
+    when_payment_is_processed_ship_order_policy       when_order_is_shipped_sit_and_relax_policy
+    """
+    history = []
+    dispatcher = InMemoryEventDispatcher()
+    mono_module = MonoModule(domain_event_dispatcher=dispatcher, history=history)
+
+    with mono_module.unit_of_work():
+        mono_module.execute_command(CompleteOrderCommand(order_id="order1"))
+
+    assert history == [
+        "completing order1",
+        "starting when_order_is_completed_process_payment_policy for order1",
+        "processing payment for order1",
+        "starting when_payment_is_processed_open_champagne_policy for order1",
+        "starting when_order_is_completed_ship_order_policy for order1",
+        "shipping order1",
+        "starting when_order_is_shipped_sit_and_relax_policy for order1",
+    ]