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-v2.12.0-blue)
+![Current Release](https://img.shields.io/badge/release-v2.12.2-blue)
 [![codecov](https://codecov.io/gh/owasp-sbot/OSBot-Utils/graph/badge.svg?token=GNVW0COX1N)](https://codecov.io/gh/owasp-sbot/OSBot-Utils)
 
 

docs/dev/Python-code-formatting-guidelines.md

@@ -74,9 +74,9 @@ class SomeClass:
 
 # This style - related elements together
 class SomeClass:
-    def __init__(self, param_one,
-                       param_two )\
-              -> None:
+    def __init__(self, param_one ,
+                       param_two  
+                )               -> None:
         self.param_one = param_one
         self.param_two = param_two
 ```
@@ -146,10 +146,10 @@ from osbot_utils.helpers.Safe_Id                     import Safe_Id
 ### Method Signature Layout
 
 ```python
-def method_name(self, first_param  : Type1         ,                               # Method purpose comment
-                      second_param : Type2         ,                               # Aligned at column 80
-                      third_param  : Type3 = None  )\                              # Default values align with type
-            -> ReturnType:                                                         # Return on new line
+def method_name(self, first_param  : Type1        ,                               # Method purpose comment
+                      second_param : Type2        ,                               # Aligned at column 80
+                      third_param  : Type3 = None                                 # Default values align with type
+               )                  -> ReturnType:                                  # Return on new line
 ```
 
 Key aspects:
@@ -165,11 +165,11 @@ Key aspects:
 ### Parameter Documentation
 
 ```python
-def complex_operation(self, data_input    : Dict[str, Any]   ,                   # Primary data structure
-                           config_options : Optional[Config] ,                   # Processing configuration 
-                           max_retries    : int = 3          ,                   # Maximum retry attempts
-                           timeout_ms     : float = 1000.0   )\                  # Operation timeout
-                  -> Tuple[Results, Metrics]:                                    # Returns results and metrics
+def complex_operation(self, data_input     : Dict    [str, Any]         ,         # Primary data structure
+                            config_options : Optional[Config  ]         ,         # Processing configuration 
+                            max_retries    : int                = 3     ,         # Maximum retry attempts
+                            timeout_ms     : float              = 1000.0          # Operation timeout
+                      )                   -> Tuple[Results, Metrics]:             # Returns results and metrics
 ```
 
 Guidelines:
@@ -185,27 +185,27 @@ Methods should be grouped by functionality with clear separation:
 
 ```python
     # Core initialization methods
-    def __init__(self, config: Config )\                                         # Initialize with configuration
-              -> None:                                                           
+    def __init__(self, config : Config                                           # Initialize with configuration
+                )            -> None:                                                           
 
-    def setup(self, options: Dict[str, Any] )\                                   # Configure processing options
-           -> bool:                                                              
+    def setup(self, options: Dict[str, Any]                                      # Configure processing options
+             )            -> bool:                                                              
 
 
     # Data validation methods  
-    def validate_input(self, data        : InputData    ,                        # Validate input format
-                             strict_mode : bool = False )\                       # Enable strict validation
-                    -> ValidationResult:
+    def validate_input(self, data        : InputData        ,                    # Validate input format
+                             strict_mode : bool      = False                     # Enable strict validation
+                      )                 -> ValidationResult:
 
     def validate_output(self, result     : OutputData ,                          # Validate output format
-                              thresholds : Thresholds )\                         # Validation thresholds
-                     -> bool:
+                              thresholds : Thresholds                            # Validation thresholds
+                      )                 -> bool:
 
 
     # Processing methods
     def process_item(self, item     : DataItem ,                                 # Process single data item
-                           settings : Settings )\                                # Processing settings
-                  -> ProcessedItem:
+                           settings : Settings                                   # Processing settings
+                    )              -> ProcessedItem:
 ```
 Note how the return type name assigns with the variable self, and there is always at least one space before the : and the ,
 
@@ -218,8 +218,8 @@ def process_batch(self, items           : List[DataItem]                   ,
                        batch_config     : BatchConfig                      ,     # Batch processing config
                        error_handler    : ErrorHandler                     ,     # Handles processing errors
                        retry_strategy   : Optional[Strategy]               ,     # Retry strategy to use
-                       metrics_callback : Callable[[Metrics], None] = None )\    # Metrics reporting callback
-                 -> BatchResults:                                                # Processed batch results
+                       metrics_callback : Callable[[Metrics], None] = None       # Metrics reporting callback
+                  )                    -> BatchResults:                          # Processed batch results
 ```
 
 Guidelines:

osbot_utils/helpers/Obj_Id.py

@@ -0,0 +1,29 @@
+import random
+
+_hex_table = [f"{i:02x}" for i in range(256)]
+
+def is_obj_id(value: str):
+    var_type = type(value)
+    if var_type is Obj_Id:
+        return True
+    if var_type is str:
+        if len(value) == 8:         # todo: add efficient check if we only have hex values
+            return True
+    return False
+
+def new_obj_id():
+    return hex(random.getrandbits(32))[2:].zfill(8)  # slice off '0x' and pad
+
+class Obj_Id(str):
+    def __new__(cls, value: str=None):
+        if value:
+            if is_obj_id(value):
+                obj_id = value
+            else:
+                raise ValueError(f'in Obj_Id: value provided was not a valid Obj_Id: {value}')
+        else:
+            obj_id = new_obj_id()
+        return super().__new__(cls, obj_id)                                          # Return a new instance of Guid initialized with the string version of the UUID
+
+    def __str__(self):
+        return self

osbot_utils/testing/performance/Performance_Measure__Session.py

@@ -11,6 +11,7 @@
 class Performance_Measure__Session(Type_Safe):
     result        : Model__Performance_Measure__Result = None                                          # Current measurement result
     assert_enabled: bool = True
+    padding       : int  = 30
 
     def calculate_raw_score(self, times: List[int]) -> int:                                     # Calculate raw performance score
         if len(times) < 3:                                                                      # Need at least 3 values for stability
@@ -89,11 +90,11 @@ def print_measurement(self, measurement: Model__Performance_Measure__Measurement
         print(f"Median  : {measurement.median_time:,}ns")
         print(f"StdDev  : {measurement.stddev_time:,.2f}ns")
 
-    def print(self, padding=12 ):                                                # Print measurement results
+    def print(self):                                                # Print measurement results
         if not self.result:
             print("No measurements taken yet")
             return
-        print(f"{self.result.name:{padding}} | score: {self.result.final_score:7,d} ns  | raw: {self.result.raw_score:7,d} ns")          # Print name and normalized score
+        print(f"{self.result.name:{self.padding}} | score: {self.result.final_score:7,d} ns  | raw: {self.result.raw_score:7,d} ns")          # Print name and normalized score
 
         return self
 
@@ -105,4 +106,23 @@ def assert_time(self, *expected_time: int):
             new_expected_time   = last_expected_time * 5                                    # using last_expected_time * 5 as the upper limit (since these tests are significantly slowed in GitHUb Actions)
             assert last_expected_time <=  self.result.final_score <= new_expected_time, f"Performance changed for {self.result.name}: expected {last_expected_time} < {self.result.final_score:,d}ns, expected {new_expected_time}"
         else:
-            assert self.result.final_score in expected_time,  f"Performance changed for {self.result.name}: got {self.result.final_score:,d}ns, expected {expected_time}"
+            assert self.result.final_score in expected_time,  f"Performance changed for {self.result.name}: got {self.result.final_score:,d}ns, expected {expected_time}"
+
+    def assert_time(self, *expected_time: int):                                              # Assert that the final score matches the expected normalized time"""
+        if self.assert_enabled is False:
+            return
+        if in_github_action():
+            last_expected_time = expected_time[-1] + 100                                    # +100 in case it is 0
+            new_expected_time   = last_expected_time * 5                                    # using last_expected_time * 5 as the upper limit (since these tests are significantly slowed in GitHUb Actions)
+            assert last_expected_time <=  self.result.final_score <= new_expected_time, f"Performance changed for {self.result.name}: expected {last_expected_time} < {self.result.final_score:,d}ns, expected {new_expected_time}"
+        else:
+            assert self.result.final_score in expected_time,  f"Performance changed for {self.result.name}: got {self.result.final_score:,d}ns, expected {expected_time}"
+
+    def assert_time__less_than(self, max_time: int):                                              # Assert that the final score matches the expected normalized time"""
+        if self.assert_enabled is False:
+            return
+        if in_github_action():
+            max_time   = max_time * 5               # adjust for GitHub's slowness
+
+        assert self.result.final_score <= max_time,  f"Performance changed for {self.result.name}: got {self.result.final_score:,d}ns, expected less than {max_time}ns"
+

osbot_utils/type_safe/shared/Type_Safe__Convert.py

@@ -20,8 +20,9 @@ def convert_to_value_from_obj_annotation(self, target, attr_name, value):
         from osbot_utils.helpers.Random_Guid    import Random_Guid
         from osbot_utils.helpers.Safe_Id        import Safe_Id
         from osbot_utils.helpers.Str_ASCII      import Str_ASCII
+        from osbot_utils.helpers.Obj_Id         import Obj_Id
 
-        TYPE_SAFE__CONVERT_VALUE__SUPPORTED_TYPES = [Guid, Random_Guid, Safe_Id, Str_ASCII, Timestamp_Now]
+        TYPE_SAFE__CONVERT_VALUE__SUPPORTED_TYPES = [Guid, Random_Guid, Safe_Id, Str_ASCII, Timestamp_Now, Obj_Id]
 
         if target is not None and attr_name is not None:
             if hasattr(target, '__annotations__'):

osbot_utils/type_safe/steps/Type_Safe__Step__Class_Kwargs.py

@@ -1,4 +1,6 @@
 from typing                                                          import Dict, Any, Type
+
+from osbot_utils.helpers.Obj_Id import Obj_Id
 from osbot_utils.helpers.Random_Guid                                 import Random_Guid
 from osbot_utils.type_safe.shared.Type_Safe__Cache                   import Type_Safe__Cache, type_safe_cache
 from osbot_utils.type_safe.shared.Type_Safe__Shared__Variables       import IMMUTABLE_TYPES
@@ -43,8 +45,10 @@ def is_kwargs_cacheable(self, cls, kwargs: Dict[str, Any]) -> bool:
         match      = all(isinstance(value, IMMUTABLE_TYPES) for value in kwargs.values())
 
         if match:                                                                       # check for special cases that we can't cache (like Random_Guid)
-            if Random_Guid in list(dict(annotations).values()):                         # todo: need to add the other special cases (like Timestamp_Now)
-
+            annotations_types = list(dict(annotations).values())
+            if Random_Guid in annotations_types:                         # todo: need to add the other special cases (like Timestamp_Now)
+                return False
+            if Obj_Id in annotations_types:                             # we can't cache Obj_id, since this would give us the same ID everutime
                 return False
         return match
 

osbot_utils/version

@@ -1 +1 @@
-v2.12.0
+v2.12.2

pyproject.toml

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

tests/unit/helpers/test_Obj_Id.py

@@ -0,0 +1,27 @@
+from unittest import TestCase
+
+from osbot_utils.helpers.Obj_Id import new_obj_id, is_obj_id, Obj_Id
+from osbot_utils.testing.performance.Performance_Measure__Session import Performance_Measure__Session
+from osbot_utils.utils.Dev import pprint
+
+
+class test_Obj_Id(TestCase):
+
+    def test__new__(self):
+        obj_id = Obj_Id()
+        assert type(obj_id           ) is Obj_Id
+        assert isinstance(obj_id, str) is True
+        assert is_obj_id(obj_id      ) is True
+
+    def test_new_obj_id(self):
+        obj_id = new_obj_id()
+        assert is_obj_id(obj_id)
+
+    def test__perf__new__(self):
+        with Performance_Measure__Session() as _:
+            _.measure(lambda: Obj_Id()    ).assert_time__less_than(600)
+
+    def test__perf__new_obj_id(self):
+        with Performance_Measure__Session() as _:
+            _.measure(lambda: new_obj_id()    ).assert_time__less_than(300)
+            _.measure(lambda: is_obj_id('abc')).assert_time__less_than(300)