Merge dev into main

README.md

@@ -2,7 +2,7 @@
 
 Powerful Python util methods and classes that simplify common apis and tasks.
 
-![Current Release](https://img.shields.io/badge/release-v1.81.0-blue)
+![Current Release](https://img.shields.io/badge/release-v1.81.2-blue)
 [![codecov](https://codecov.io/gh/owasp-sbot/OSBot-Utils/graph/badge.svg?token=GNVW0COX1N)](https://codecov.io/gh/owasp-sbot/OSBot-Utils)
 
 

osbot_utils/base_classes/Type_Safe.py

@@ -361,3 +361,5 @@ def serialize_to_dict(obj):
         return data
     else:
         raise TypeError(f"Type {type(obj)} not serializable")
+        #return f"UNSERIALIZABLE({type(obj).__name__})"             # todo: see if there are valid use cases for this
+

osbot_utils/helpers/generators/Generator_Manager.py

@@ -8,6 +8,7 @@
 from osbot_utils.helpers.generators.Generator_Context_Manager   import Generator_Context_Manager
 from osbot_utils.helpers.generators.Model__Generator_State      import Model__Generator_State
 from osbot_utils.helpers.generators.Model__Generator_Target     import Model__Generator_Target
+from osbot_utils.utils.Lists                                    import list_group_by
 
 
 class Generator_Manager(Type_Safe):                                                                        # Class for managing multiple generator targets
@@ -92,6 +93,21 @@ def stop_all(self) -> int:
                     stopped_count += 1                                                                     # Increment the stopped count if successful
             return stopped_count                                                                           # Return the total number of stopped generators
 
+    def status(self):
+        items = []
+        for _, generator in self.generators.items():
+            item = dict(target_method_name = generator.target.__name__,
+                        target_state       = generator.state.value,
+                        target_id          = generator.target_id )
+            items.append(item)
+        items__by_state = list_group_by(items, 'target_state')
+        result = {}
+        for state, items in items__by_state.items():
+            result[state] = len(items)
+        result['data'] = items__by_state
+        return result
+
+
     def target_id(self, target: GeneratorType) -> Union[Random_Guid, None]:                                # Method to get the ID of a specific generator
         with self.lock:                                                                                    # Acquire the lock for thread-safe access
             for generator in list(self.generators.values()):                                               # Iterate over the generator targets

osbot_utils/version

@@ -1 +1 @@
-v1.81.0
+v1.81.2

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name        = "osbot_utils"
-version     = "v1.81.0"
+version     = "v1.81.2"
 description = "OWASP Security Bot - Utils"
 authors     = ["Dinis Cruz <[email protected]>"]
 license     = "MIT"

tests/unit/helpers/generators/test_Generator_Manager.py

@@ -38,6 +38,12 @@ def sample_generator():
         self.assertEqual(generator.target, gen)
         self.assertEqual(generator.state, Model__Generator_State.RUNNING)
 
+
+        assert self.manager.status() == { 'data': { 'running': [{ 'target_id'         : target_id        ,
+                                                                  'target_state'      : 'running'         ,
+                                                                  'target_method_name': 'sample_generator'}]},
+                                         'running': 1}
+
     def test_add_duplicate_generator(self):
         def sample_generator():
             yield 1