Multiple changes to Flow and Task helpers

Added Flow__Events main class
Refactored Flow related model objects into a models folder

osbot_utils/helpers/flows/Flow.py

@@ -2,14 +2,15 @@
 import logging
 import typing
 
-from osbot_utils.base_classes.Type_Safe import Type_Safe
-from osbot_utils.helpers.CFormat import CFormat, f_dark_grey, f_magenta, f_bold
-from osbot_utils.helpers.flows.Flow__Config import Flow__Config
-from osbot_utils.testing.Stdout         import Stdout
-from osbot_utils.utils.Misc             import random_id, lower
-from osbot_utils.utils.Python_Logger    import Python_Logger
-from osbot_utils.utils.Str              import ansis_to_texts
-from osbot_utils.utils.Threads import invoke_async, invoke_in_new_event_loop
+from osbot_utils.base_classes.Type_Safe             import Type_Safe
+from osbot_utils.helpers.CFormat                    import CFormat, f_dark_grey, f_magenta, f_bold
+from osbot_utils.helpers.flows.models.Flow__Config  import Flow__Config
+from osbot_utils.helpers.flows.Flow__Events         import flow_events
+from osbot_utils.testing.Stdout                     import Stdout
+from osbot_utils.utils.Misc                         import random_id, lower
+from osbot_utils.utils.Python_Logger                import Python_Logger
+from osbot_utils.utils.Str                          import ansis_to_texts
+from osbot_utils.utils.Threads                      import invoke_in_new_event_loop
 
 FLOW__RANDOM_ID__PREFIX    = 'flow_id__'
 FLOW__RANDOM_NAME__PREFIX  = 'flow_name__'
@@ -54,6 +55,7 @@ def execute(self):
         return self.execute_flow()
 
     def execute_flow(self):
+        flow_events.on__flow__start(self)
         if self.flow_config.log_to_memory:
             self.logger.add_memory_logger()                                     # todo: move to method that does pre-execute tasks
 
@@ -72,6 +74,7 @@ def execute_flow(self):
         if self.flow_config.log_to_memory:
             self.captured_exec_logs = self.log_messages_with_colors()
             self.logger.remove_memory_logger()                                                          # todo: move to method that does post-execute tasks
+        flow_events.on__flow__stop(self)
         return self
 
     def f__flow_id(self):
@@ -170,6 +173,3 @@ def setup_flow_run(self):
         with self as _:
             if not _.flow_id:
                 _.flow_id = self.random_flow_id()
-            #if not _.flow_name:
-            #    _.flow_name = self.flow_target.__name__
-                #self.random_flow_name()

osbot_utils/helpers/flows/Flow__Events.py

@@ -0,0 +1,32 @@
+from osbot_utils.base_classes.Type_Safe                 import Type_Safe
+from osbot_utils.helpers.flows.models.Flow__Event       import Flow__Event
+from osbot_utils.helpers.flows.models.Flow__Event_Type  import Flow__Event_Type
+
+
+class Flow_Events(Type_Safe):
+    event_listeners : list
+
+    def on__flow__start(self, flow):
+        flow_event = Flow__Event(event_type=Flow__Event_Type.FLOW_START, event_source=flow)
+        self.raise_event(flow_event)
+
+    def on__flow__stop(self, flow):                                                         # todo: see of flow_ended or flow_completed are better names
+        flow_event = Flow__Event(event_type=Flow__Event_Type.FLOW_STOP , event_source=flow)
+        self.raise_event(flow_event)
+
+    def on__task__start(self, task):
+        flow_event = Flow__Event(event_type=Flow__Event_Type.TASK_START, event_source=task)
+        self.raise_event(flow_event)
+
+    def on__task__stop(self, task):                                                         # todo: see of flow_ended or flow_completed are better names
+        flow_event = Flow__Event(event_type=Flow__Event_Type.TASK_STOP , event_source=task)
+        self.raise_event(flow_event)
+
+    def raise_event(self, flow_event):
+        for listener in self.event_listeners:
+            try:
+                listener(flow_event)
+            except Exception as error:
+                print(f"Error in listener: {error}")
+
+flow_events = Flow_Events()

osbot_utils/helpers/flows/Task.py

@@ -2,18 +2,22 @@
 import inspect
 import typing
 
+from osbot_utils.utils.Misc import random_id, lower
+
 from osbot_utils.helpers.Dependency_Manager import Dependency_Manager
+from osbot_utils.helpers.flows.Flow__Events import flow_events
 from osbot_utils.utils.Dev import pprint
 
 from osbot_utils.testing.Stdout             import Stdout
 from osbot_utils.helpers.CFormat            import CFormat, f_dark_grey, f_red, f_blue, f_bold
 from osbot_utils.base_classes.Type_Safe     import Type_Safe
 from osbot_utils.helpers.flows.Flow         import Flow
 
+TASK__RANDOM_ID__PREFIX    = 'task_id__'
 
 class Task(Type_Safe):
     data                : dict                          # dict available to the task to add and collect data
-    task_id             : str                           # todo add a random Id value to this
+    task_id             : str
     task_name           : str                           # make this the function mame
     cformat             : CFormat
     resolved_args       : tuple
@@ -44,6 +48,9 @@ def execute__before(self):
         if not self.task_name and self.task_target:
             self.task_name = self.task_target.__name__
 
+        if not self.task_id:
+            self.task_id = self.random_task_id()
+
         self.task_flow.executed_tasks.append(self)
         self.task_flow.logger.debug(f"Executing task '{f_blue(self.task_name)}'")
         dependency_manager = Dependency_Manager()
@@ -52,6 +59,7 @@ def execute__before(self):
         dependency_manager.add_dependency('task_data', self.data          )
         dependency_manager.add_dependency('flow_data', self.task_flow.data)
         self.resolved_args, self.resolved_kwargs = dependency_manager.resolve_dependencies(self.task_target, *self.task_args, **self.task_kwargs)
+        flow_events.on__task__start(self)
 
     def execute__task_target__sync(self):
         try:
@@ -78,6 +86,8 @@ def execute__after(self):
                 raise Exception(f"'{self.task_name}' failed and task raise_on_error was set to True. Stopping flow execution")
 
         self.print_task_finished_message()
+
+        flow_events.on__task__stop(self)
         return self.task_return_value
 
 
@@ -101,3 +111,5 @@ def print_task_return_value(self):
         self.task_flow.logger.debug(f"{f_dark_grey('Task return value')}: {f_bold(self.task_return_value)}")
 
 
+    def random_task_id(self):
+        return lower(random_id(prefix=TASK__RANDOM_ID__PREFIX))

osbot_utils/helpers/flows/models/Flow__Event.py

@@ -0,0 +1,6 @@
+from osbot_utils.base_classes.Type_Safe import Type_Safe
+
+class Flow__Event(Type_Safe):
+    event_type  : str
+    event_source: object
+    event_data  : dict

osbot_utils/helpers/flows/models/Flow__Event_Type.py

@@ -0,0 +1,8 @@
+from osbot_utils.base_classes.Type_Safe import Type_Safe
+
+
+class Flow__Event_Type(Type_Safe):
+    FLOW_START : str = 'flow_start'
+    FLOW_STOP  : str = 'flow_stop'
+    TASK_START : str = 'task_start'
+    TASK_STOP  : str = 'task_stop'

tests/unit/helpers/flows/test_Flow__Events.py

@@ -0,0 +1,38 @@
+from unittest                                   import TestCase
+from osbot_utils.helpers.flows.Flow             import Flow
+from osbot_utils.helpers.flows.decorators.flow  import flow
+from osbot_utils.helpers.flows.Flow__Events     import Flow__Event_Type, Flow_Events, flow_events
+
+
+class test_Flow__Events(TestCase):
+
+    def setUp(self):
+        self.flow_events = Flow_Events()
+
+    def test_on__flow__start(self):
+        def event_listener(flow_event):
+            assert flow_event.event_type == Flow__Event_Type.FLOW_START
+            assert flow_event.event_source == 'flow'
+            assert flow_event.event_data == {}
+        self.flow_events.event_listeners.append(event_listener)
+        self.flow_events.on__flow__start('flow')
+
+    def test_global_flow_events(self):
+        def event_listener(flow_event):
+            assert flow_event.event_type         == Flow__Event_Type.FLOW_START
+            assert flow_event.event_type         == Flow__Event_Type.FLOW_START
+            assert type(flow_event.event_source) is Flow
+
+        @flow()
+        def an_flow() -> Flow:
+            print('inside the flow')
+
+        flow_events.event_listeners.append(event_listener)
+        assert event_listener in flow_events.event_listeners
+
+        an_flow().execute_flow()
+
+        flow_events.event_listeners.remove(event_listener)
+        assert event_listener not in flow_events.event_listeners
+
+

tests/unit/helpers/flows/test_decorator__flow.py

@@ -2,7 +2,7 @@
 from osbot_utils.base_classes.Type_Safe                 import Type_Safe
 from osbot_utils.context_managers.disable_root_loggers  import disable_root_loggers
 from osbot_utils.helpers.flows.Flow                     import Flow
-from osbot_utils.helpers.flows.Flow__Config import Flow__Config
+from osbot_utils.helpers.flows.models.Flow__Config      import Flow__Config
 from osbot_utils.helpers.flows.Task                     import Task
 from osbot_utils.helpers.flows.decorators.flow          import flow
 from osbot_utils.helpers.flows.decorators.task          import task