check for return type from task handlers

Author: pgorecki

src/modules/bidding/application/event.py

@@ -5,12 +5,13 @@
 from modules.bidding.domain.repositories import ListingRepository
 from modules.bidding.domain.value_objects import Seller
 from modules.catalog.domain.events import ListingPublishedEvent
+from seedwork.application import EventResult
 
 
 @bidding_module.domain_event_handler
 def when_listing_is_published_start_auction(
     event: ListingPublishedEvent, listing_repository: ListingRepository
-):
+) -> EventResult:
     listing = Listing(
         id=event.listing_id,
         seller=Seller(id=event.seller_id),
@@ -19,3 +20,4 @@ def when_listing_is_published_start_auction(
         ends_at=datetime.now() + timedelta(days=7),
     )
     listing_repository.add(listing)
+    return EventResult.success()

src/modules/bidding/tests/application/test_create_listing_when_draft_is_published.py

@@ -1,26 +1,24 @@
 import pytest
-from sqlalchemy.orm import Session
 
+from modules.bidding.domain.repositories import (
+    ListingRepository as BiddingListingRepository,
+)
 from modules.catalog.domain.events import ListingPublishedEvent
 from seedwork.domain.value_objects import UUID, Money
 
 
-@pytest.mark.skip  # this test needs to be fixed
 @pytest.mark.integration
-def test_create_listing_on_draft_published_event(engine):
-    module = BiddingModule()
-    listing_id = UUID.v4()
-    with Session(engine) as db_session:
-        with module.unit_of_work(db_session=db_session) as uow:
-            module.handle_domain_event(
-                ListingPublishedEvent(
-                    listing_id=listing_id,
-                    seller_id=UUID.v4(),
-                    ask_price=Money(10),
-                )
+def test_create_listing_on_draft_published_event(app, engine):
+    listing_id = UUID(int=1)
+    with app.transaction_context() as ctx:
+        ctx.handle_domain_event(
+            ListingPublishedEvent(
+                listing_id=listing_id,
+                seller_id=UUID.v4(),
+                ask_price=Money(10),
             )
-        db_session.commit()
+        )
 
-    with Session(engine) as db_session:
-        with module.unit_of_work(db_session=db_session) as uow:
-            assert uow.listing_repository.count() == 1
+    with app.transaction_context() as ctx:
+        listing_repository = ctx.get_service(BiddingListingRepository)
+        assert listing_repository.count() == 1

src/seedwork/application/__init__.py

@@ -117,7 +117,7 @@ def _wrap_with_middlewares(
             p = partial(middleware, self, p, command, query, event)
         return p
 
-    def execute_query(self, query):
+    def execute_query(self, query) -> QueryResult:
         assert (
             self.task is None
         ), "Cannot execute query while another task is being executed"
@@ -127,12 +127,15 @@ def execute_query(self, query):
         handler_kwargs = self.dependency_provider.get_handler_kwargs(
             handler_func, **self.overrides
         )
-        handler_func = partial(handler_func, query, **handler_kwargs)
-        wrapped_handler = self._wrap_with_middlewares(handler_func, query=query)
+        p = partial(handler_func, query, **handler_kwargs)
+        wrapped_handler = self._wrap_with_middlewares(p, query=query)
         result = wrapped_handler()
+        assert isinstance(
+            result, QueryResult
+        ), f"Got {result} instead of QueryResult from {handler_func}"
         return result
 
-    def execute_command(self, command):
+    def execute_command(self, command) -> CommandResult:
         assert (
             self.task is None
         ), "Cannot execute command while another task is being executed"
@@ -142,53 +145,55 @@ def execute_command(self, command):
         handler_kwargs = self.dependency_provider.get_handler_kwargs(
             handler_func, **self.overrides
         )
-        handler_func = partial(handler_func, command, **handler_kwargs)
-        wrapped_handler = self._wrap_with_middlewares(handler_func, command=command)
+        p = partial(handler_func, command, **handler_kwargs)
+        wrapped_handler = self._wrap_with_middlewares(p, command=command)
 
         # execute wrapped command handler
         command_result = wrapped_handler()
+        assert isinstance(
+            command_result, CommandResult
+        ), f"Got {command_result} instead of CommandResult from {handler_func}"
 
         self.next_commands = []
         self.integration_events = []
         event_queue = command_result.events.copy()
         while len(event_queue) > 0:
             event = event_queue.pop(0)
             if isinstance(event, IntegrationEvent):
-                self._process_integration_event(event)
+                self.collect_integration_event(event)
 
             elif isinstance(event, DomainEvent):
-                new_command, new_events = self._process_domain_event(event)
-                self.next_commands.extend(new_command)
-                event_queue.extend(new_events)
+                event_results = self.handle_domain_event(event)
+                self.next_commands.extend(event_results.commands)
+                event_queue.extend(event_results.events)
 
         return CommandResult.success(payload=command_result.payload)
 
+    def handle_domain_event(self, event) -> EventResultSet:
+        event_results = []
+        for handler_func in self.app.get_event_handlers(event):
+            handler_kwargs = self.dependency_provider.get_handler_kwargs(
+                handler_func, **self.overrides
+            )
+            p = partial(handler_func, event, **handler_kwargs)
+            wrapped_handler = self._wrap_with_middlewares(p, event=event)
+            result = wrapped_handler()
+            assert isinstance(
+                result, EventResult
+            ), f"Got {result} instead of EventResult from {handler_func}"
+            event_results.append(result)
+        return EventResultSet(event_results)
+
+    def collect_integration_event(self, event):
+        self.integration_events.append(event)
+
     def get_service(self, service_cls):
         return self.dependency_provider.get_dependency(service_cls)
 
     @property
     def current_user(self):
         return self.dependency_provider.get_dependency("current_user")
 
-    def _process_integration_event(self, event):
-        self.integration_events.append(event)
-
-    def _process_domain_event(self, event):
-        new_commands = []
-        new_events = []
-        for handler_func in self.app.get_event_handlers(event):
-            handler_kwargs = self.dependency_provider.get_handler_kwargs(
-                handler_func, **self.overrides
-            )
-            event_handler = partial(handler_func, event, **handler_kwargs)
-            wrapped_handler = self._wrap_with_middlewares(event_handler, event=event)
-            result = wrapped_handler()
-            if isinstance(result, Command):
-                new_commands.append(result)
-            elif isinstance(result, EventResult):
-                new_events.extend(result.events)
-        return new_commands, new_events
-
 
 class ApplicationModule:
     def __init__(self, name, version=1.0):

src/seedwork/application/events.py

@@ -24,6 +24,7 @@ class EventResult:
 
     event_id: UUID = None
     payload: Any = None
+    command: Any = None  # command th
     events: list[DomainEvent] = field(default_factory=list)
     errors: list[Any] = field(default_factory=list)
 
@@ -48,14 +49,14 @@ def failure(cls, message="Failure", exception=None) -> "CommandResult":
 
     @classmethod
     def success(
-        cls, event_id=None, payload=None, event=None, events=None
+        cls, event_id=None, payload=None, command=None, event=None, events=None
     ) -> "EventResult":
         """Creates a successful result"""
         if events is None:
             events = []
         if event:
             events.append(event)
-        return cls(event_id=event_id, payload=payload, events=events)
+        return cls(event_id=event_id, payload=payload, command=command, events=events)
 
 
 class EventResultSet(set):
@@ -70,3 +71,8 @@ def events(self):
         for event in self:
             all_events.extend(event.events)
         return all_events
+
+    @property
+    def commands(self):
+        all_commands = [event.command for event in self if event.command]
+        return all_commands