added get_origin to Type_Safe__Not_Cache

added misses stats to Type_Save__Cache print function
more misc fixes

osbot_utils/type_safe/shared/Type_Safe__Annotations.py

@@ -5,7 +5,7 @@
 class Type_Safe__Annotations:
 
     def all_annotations(self, target):
-        return type_safe_cache.get_annotations(target)                          # use cache
+        return type_safe_cache.get_obj_annotations(target)                          # use cache
 
     def all_annotations__in_class(self, cls):
         return type_safe_cache.get_class_annotations(cls)
@@ -23,5 +23,8 @@ def obj_is_attribute_annotation_of_type(self, target, attr_name, expected_type):
             return True
         return False
 
+    def get_origin(self, var_type):
+        return type_safe_cache.get_origin(var_type)
+
+type_safe_annotations = Type_Safe__Annotations()
 
-type_safe_annotations = Type_Safe__Annotations()

osbot_utils/type_safe/shared/Type_Safe__Cache.py

@@ -1,94 +1,108 @@
 import inspect
-from typing                                                     import get_origin
 from weakref                                                    import WeakKeyDictionary
 from osbot_utils.type_safe.shared.Type_Safe__Not_Cached         import type_safe_not_cached
 from osbot_utils.type_safe.shared.Type_Safe__Shared__Variables  import IMMUTABLE_TYPES
 
 
 class Type_Safe__Cache:
 
-    _cls__annotations_cache : WeakKeyDictionary
-    _cls__immutable_vars    : WeakKeyDictionary
-    _cls__kwargs_cache      : WeakKeyDictionary
-    _obj__annotations_cache : WeakKeyDictionary
-    _get_origin_cache       : WeakKeyDictionary
-    _mro_cache              : WeakKeyDictionary
-    _valid_vars_cache       : WeakKeyDictionary
-
-    cache_hit__cls__annotations   : int  = 0
-    cache_hit__cls__kwargs        : int  = 0
-    cache_hit__cls__immutable_vars: int  = 0
-    cache_hit__obj__annotations   : int  = 0
-    cache_hit__get_origin         : int  = 0
-    cache_hit__mro                : int  = 0
-    cache_hit__valid_vars         : int  = 0
-    skip_cache                    : bool = False
+    _cls__annotations_cache  : WeakKeyDictionary
+    _cls__immutable_vars     : WeakKeyDictionary
+    _cls__kwargs_cache       : WeakKeyDictionary
+    _obj__annotations_cache  : WeakKeyDictionary
+    _type__get_origin_cache  : WeakKeyDictionary
+    _mro_cache               : WeakKeyDictionary
+    _valid_vars_cache        : WeakKeyDictionary
+
+    cache__miss__cls__annotations   : int = 0
+    cache__miss__cls__kwargs        : int = 0
+    cache__miss__cls__immutable_vars: int = 0
+    cache__miss__obj__annotations   : int = 0
+    cache__miss__type__get_origin   : int = 0
+    cache__miss__mro                : int = 0
+    cache__miss__valid_vars         : int = 0
+
+    cache__hit__cls__annotations    : int  = 0
+    cache__hit__cls__kwargs         : int  = 0
+    cache__hit__cls__immutable_vars : int  = 0
+    cache__hit__obj__annotations    : int  = 0
+    cache__hit__type__get_origin    : int  = 0
+    cache__hit__mro                 : int  = 0
+    cache__hit__valid_vars          : int  = 0
+    skip_cache                      : bool = False
 
 
     # Caching system for Type_Safe methods
     def __init__(self):
-        self._cls__annotations_cache = WeakKeyDictionary()                                        # Cache for class annotations
-        self._cls__immutable_vars    = WeakKeyDictionary()                                        # Cache for class immutable vars
-        self._cls__kwargs_cache      = WeakKeyDictionary()                                        # Cache for class kwargs
-        self._obj__annotations_cache = WeakKeyDictionary()                                        # Cache for object annotations
-        self._get_origin_cache       = WeakKeyDictionary()                                        # Cache for get_origin results
-        self._mro_cache              = WeakKeyDictionary()                                        # Cache for Method Resolution Order
-        self._valid_vars_cache       = WeakKeyDictionary()
+        self._cls__annotations_cache  = WeakKeyDictionary()                                        # Cache for class annotations
+        self._cls__immutable_vars     = WeakKeyDictionary()                                        # Cache for class immutable vars
+        self._cls__kwargs_cache       = WeakKeyDictionary()                                        # Cache for class kwargs
+        self._obj__annotations_cache  = WeakKeyDictionary()                                        # Cache for object annotations
+        self._type__get_origin_cache  = WeakKeyDictionary()                                        # Cache for tp (type) get_origin results
+        self._mro_cache               = WeakKeyDictionary()                                        # Cache for Method Resolution Order
+        self._valid_vars_cache        = WeakKeyDictionary()
 
     def get_cls_kwargs(self, cls):
-        if self.skip_cache or cls not in self._cls__kwargs_cache:
-            return None
+        cls_kwargs = self._cls__kwargs_cache.get(cls)
+
+        if cls_kwargs is None:
+            self.cache__miss__cls__kwargs += 1
         else:
-            self.cache_hit__cls__kwargs += 1
-        return self._cls__kwargs_cache.get(cls)
+            self.cache__hit__cls__kwargs += 1
+        return cls_kwargs
 
-    def get_annotations(self, target):
+    def get_obj_annotations(self, target):
         if target is None:
             return {}
         annotations_key = target.__class__
         annotations = self._obj__annotations_cache.get(annotations_key)                          # this is a more efficient cache retrieval pattern (we only get the data from the dict once)
-        if not annotations:                                                     # todo: apply this to the other cache getters
-            if self.skip_cache or annotations_key not in self._obj__annotations_cache:
-                annotations = dict(type_safe_not_cached.all_annotations(target).items())
-                self._obj__annotations_cache[annotations_key] = annotations
+        if self.skip_cache or annotations is None:
+            annotations = dict(type_safe_not_cached.all_annotations(target).items())
+            self._obj__annotations_cache[annotations_key] = annotations
+            self.cache__miss__obj__annotations += 1
         else:
-            self.cache_hit__obj__annotations += 1
+            self.cache__hit__obj__annotations += 1
         return annotations
 
     def get_class_annotations(self, cls):
         annotations = self._cls__annotations_cache.get(cls)                          # this is a more efficient cache retrieval pattern (we only get the data from the dict once)
-        if not annotations:                                                     # todo: apply this to the other cache getters
-            if self.skip_cache or cls not in self._cls__annotations_cache:
-                annotations = type_safe_not_cached.all_annotations__in_class(cls).items()
-                self._cls__annotations_cache[cls] = annotations
+        if self.skip_cache or annotations is None:                                                     # todo: apply this to the other cache getters
+            annotations = type_safe_not_cached.all_annotations__in_class(cls).items()
+            self._cls__annotations_cache[cls] = annotations
+            self.cache__miss__cls__annotations +=1
         else:
-            self.cache_hit__cls__annotations += 1
+            self.cache__hit__cls__annotations += 1
         return annotations
 
     def get_class_immutable_vars(self, cls):
         immutable_vars = self._cls__immutable_vars.get(cls)
-        if self.skip_cache or not immutable_vars:
-            annotations                    = self.get_class_annotations(cls)
-            immutable_vars                 = {key: value for key, value in annotations if value in IMMUTABLE_TYPES}
-            self._cls__immutable_vars[cls] = immutable_vars
+        if self.skip_cache or immutable_vars is None:
+            annotations                            = self.get_class_annotations(cls)
+            immutable_vars                         = {key: value for key, value in annotations if value in IMMUTABLE_TYPES}
+            self._cls__immutable_vars[cls]         = immutable_vars
+            self.cache__miss__cls__immutable_vars += 1
         else:
-            self.cache_hit__cls__immutable_vars += 1
+            self.cache__hit__cls__immutable_vars   += 1
         return immutable_vars
 
     def get_class_mro(self, cls):
         if self.skip_cache or cls not in self._mro_cache:
-            self._mro_cache[cls] = inspect.getmro(cls)
+            self._mro_cache[cls]   = inspect.getmro(cls)
+            self.cache__miss__mro += 1
         else:
-            self.cache_hit__mro += 1
+            self.cache__hit__mro   += 1
         return self._mro_cache[cls]
 
 
     def get_origin(self, var_type):                                                             # Cache expensive get_origin calls
-        if self.skip_cache or var_type not in self._get_origin_cache:
-            self._get_origin_cache[var_type] = get_origin(var_type)
+        if self.skip_cache or var_type not in self._type__get_origin_cache:
+            origin                                 = type_safe_not_cached.get_origin(var_type)
+            self._type__get_origin_cache[var_type] = origin
+            self.cache__miss__type__get_origin    += 1
         else:
-            self.cache_hit__get_origin += 1
-        return self._get_origin_cache[var_type]
+            origin = self._type__get_origin_cache[var_type]
+            self.cache__hit__type__get_origin      += 1
+        return origin
 
     # todo: see if we have cache misses and invalid hits based on the validator (we might need more validator specific methods)
     def get_valid_class_variables(self, cls, validator):
@@ -97,25 +111,30 @@ def get_valid_class_variables(self, cls, validator):
             for name, value in vars(cls).items():
                 if not validator(name, value):
                     valid_variables[name] = value
-            self._valid_vars_cache[cls] = valid_variables
+            self._valid_vars_cache[cls]   = valid_variables
+            self.cache__miss__valid_vars += 1
         else:
-            self.cache_hit__valid_vars += 1
+            self.cache__hit__valid_vars  += 1
         return self._valid_vars_cache[cls]
 
     def set_cache__cls_kwargs(self, cls, kwargs):
-        self._cls__kwargs_cache[cls] = kwargs
+        if self.skip_cache is False:
+            self._cls__kwargs_cache[cls] = kwargs
         return kwargs
 
     def print_cache_hits(self):
         print()
         print("###### Type_Safe_Cache Hits ########")
         print()
-        print(f"  annotations        : {self.cache_hit__cls__annotations    }")
-        print(f"  cls__kwargs        : {self.cache_hit__cls__kwargs         }")
-        print(f"  cls__immutable_vars: {self.cache_hit__cls__immutable_vars }")
-        print(f"  obj__annotations   : {self.cache_hit__obj__annotations    }")
-        print(f"  get_origin         : {self.cache_hit__get_origin          }")
-        print(f"  mro                : {self.cache_hit__mro                 }")
-        print(f"  valid_vars         : {self.cache_hit__valid_vars          }")
-
-type_safe_cache = Type_Safe__Cache()
+        print( "  cache name          | hits   | miss  |  size |")
+        print( "----------------------|--------|-------|-------|")
+        print(f"  annotations         | {self.cache__hit__cls__annotations   :5}  | {self.cache__miss__cls__annotations    :5} | {len(self._obj__annotations_cache) :5} |")
+        print(f"  cls__kwargs         | {self.cache__hit__cls__kwargs        :5}  | {self.cache__miss__cls__kwargs         :5} | {len(self._cls__kwargs_cache     ) :5} |")
+        print(f"  cls__immutable_vars | {self.cache__hit__cls__immutable_vars:5}  | {self.cache__miss__cls__immutable_vars :5} | {len(self._cls__immutable_vars   ) :5} |")
+        print(f"  obj__annotations    | {self.cache__hit__obj__annotations   :5}  | {self.cache__miss__obj__annotations    :5} | {len(self._obj__annotations_cache) :5} |")
+        print(f"  type__get_origin    | {self.cache__hit__type__get_origin   :5}  | {self.cache__miss__type__get_origin    :5} | {len(self._type__get_origin_cache) :5} |")
+        print(f"  mro                 | {self.cache__hit__mro                :5}  | { self.cache__miss__mro                :5} | {len(self._mro_cache             ) :5} |")
+        print(f"  valid_vars          | {self.cache__hit__valid_vars         :5}  | {self.cache__miss__valid_vars          :5} | {len(self._valid_vars_cache      ) :5} |")
+
+type_safe_cache = Type_Safe__Cache()
+

osbot_utils/type_safe/shared/Type_Safe__Not_Cached.py

@@ -1,3 +1,5 @@
+from typing import get_origin
+
 class Type_Safe__Not_Cached:
 
     def all_annotations(self, target):
@@ -16,4 +18,7 @@ def all_annotations__in_class(self, target):
                     annotations.update(base.__annotations__)
         return annotations
 
+    def get_origin(self, var_type):
+        return get_origin(var_type)
+
 type_safe_not_cached = Type_Safe__Not_Cached()

osbot_utils/type_safe/shared/Type_Safe__Raise_Exception.py

@@ -1,4 +1,3 @@
-from typing                                                    import Any
 from osbot_utils.type_safe.shared.Type_Safe__Shared__Variables import IMMUTABLE_TYPES
 
 

osbot_utils/type_safe/shared/Type_Safe__Validation.py

@@ -1,11 +1,10 @@
 import types
 import typing
 from enum                                                           import EnumMeta
-from typing                                                         import Any, Annotated, Optional, get_args, get_origin, ForwardRef, Type, Dict
+from typing                                                         import Any, Annotated, Optional, get_args, get_origin, ForwardRef, Type, Dict, _GenericAlias
 from osbot_utils.type_safe.shared.Type_Safe__Annotations            import type_safe_annotations
 from osbot_utils.type_safe.shared.Type_Safe__Cache                  import type_safe_cache
 from osbot_utils.type_safe.shared.Type_Safe__Shared__Variables      import IMMUTABLE_TYPES
-from osbot_utils.utils.Objects                                      import obj_is_type_union_compatible
 from osbot_utils.type_safe.shared.Type_Safe__Raise_Exception        import type_safe_raise_exception
 
 
@@ -51,6 +50,21 @@ def are_types_magic_mock(self, source_type, target_type):
         #     return True
         return False
 
+    def obj_is_type_union_compatible(self, var_type, compatible_types):
+        from typing import Union
+
+        origin = get_origin(var_type)
+        if isinstance(var_type, _GenericAlias) and origin is type:              # Add handling for Type[T]
+            return type in compatible_types                                     # Allow if 'type' is in compatible types
+        if origin is Union:                                                     # For Union types, including Optionals
+            args = get_args(var_type)                                           # Get the argument types
+            for arg in args:                                                    # Iterate through each argument in the Union
+                if not (arg in compatible_types or arg is type(None)):          # Check if the argument is either in the compatible_types or is type(None)
+                    return False                                                # If any arg doesn't meet the criteria, return False immediately
+            return True                                                         # If all args are compatible, return True
+        return var_type in compatible_types or var_type is type(None)           # Check for direct compatibility or type(None) for non-Union types
+
+
     def check_if__type_matches__obj_annotation__for_union_and_annotated(self, target   : Any    ,    # Target object to check
                                                                              attr_name : str    ,    # Attribute name
                                                                              value     : Any    )\
@@ -141,7 +155,7 @@ def check_if__type_matches__obj_annotation__for_attr(self, target,
                                                                attr_name,
                                                                value
                                                          )              -> Optional[bool]:
-        annotations = type_safe_cache.get_annotations(target)
+        annotations = type_safe_cache.get_obj_annotations(target)
         attr_type   = annotations.get(attr_name)
         if attr_type:
             origin_attr_type = get_origin(attr_type)                                    # to handle when type definition contains a generic
@@ -220,7 +234,7 @@ def validate_type_compatibility(self, target      : Any             ,
     # todo: see if need to add cache support to this method     (it looks like this method is not called very often)
     def validate_type_immutability(self, var_name: str, var_type: Any) -> None:                         # Validates that type is immutable or in supported format
         if var_type not in IMMUTABLE_TYPES and var_name.startswith('__') is False:                      # if var_type is not one of the IMMUTABLE_TYPES or is an __ internal
-            if obj_is_type_union_compatible(var_type, IMMUTABLE_TYPES) is False:                        # if var_type is not something like Optional[Union[int, str]]
+            if self.obj_is_type_union_compatible(var_type, IMMUTABLE_TYPES) is False:                        # if var_type is not something like Optional[Union[int, str]]
                 if var_type not in IMMUTABLE_TYPES or type(var_type) not in IMMUTABLE_TYPES:
                     if not isinstance(var_type, EnumMeta):
                         type_safe_raise_exception.immutable_type_error(var_name, var_type)

osbot_utils/type_safe/steps/Type_Safe__Step__From_Json.py

@@ -98,7 +98,7 @@ def deserialize_type__using_value(self, value):
     def deserialize_dict__using_key_value_annotations(self, _self, key, value):
         from osbot_utils.type_safe.Type_Safe__Dict import Type_Safe__Dict
 
-        annotations            = type_safe_cache.get_annotations(_self)
+        annotations            = type_safe_cache.get_obj_annotations(_self)
         dict_annotations_tuple = get_args(annotations.get(key))
         if not dict_annotations_tuple:                                      # happens when the value is a dict/Dict with no annotations
             return value

osbot_utils/type_safe/steps/Type_Safe__Step__Set_Attr.py

@@ -73,7 +73,7 @@ def setattr(self, _super, _self, name, value):
             if self.handle_special_generic_alias(_super, _self, name, value):
                 return
 
-        annotations = dict(type_safe_cache.get_annotations(_self))
+        annotations = dict(type_safe_cache.get_obj_annotations(_self))
 
         if not annotations:                                             # can't do type safety checks if the class does not have annotations
             return _super.__setattr__(name, value)

osbot_utils/utils/Objects.py

@@ -276,23 +276,6 @@ def obj_get_value(target=None, key=None, default=None):
 def obj_values(target=None):
     return list(obj_dict(target).values())
 
-
-def obj_is_type_union_compatible(var_type, compatible_types):
-    from typing import Union
-
-    origin = get_origin(var_type)
-    if isinstance(var_type, _GenericAlias) and origin is type:              # Add handling for Type[T]
-        return type in compatible_types                                     # Allow if 'type' is in compatible types
-    if origin is Union:                                                     # For Union types, including Optionals
-        args = get_args(var_type)                                           # Get the argument types
-        for arg in args:                                                    # Iterate through each argument in the Union
-            if not (arg in compatible_types or arg is type(None)):          # Check if the argument is either in the compatible_types or is type(None)
-                return False                                                # If any arg doesn't meet the criteria, return False immediately
-        return True                                                         # If all args are compatible, return True
-    return var_type in compatible_types or var_type is type(None)           # Check for direct compatibility or type(None) for non-Union types
-
-
-
 def pickle_save_to_bytes(target: object) -> bytes:
     import pickle
     return pickle.dumps(target)