added new decorator called context (which has a before and after meth…

…od support

started adding tests for Cache_Pickle class

osbot_utils/base_classes/Cache_Pickle.py

@@ -1,39 +1,42 @@
 import os
 from functools import wraps
 
-from osbot_utils.utils.Files import path_combine, folder_create, temp_folder_current, file_exists, pickle_load_from_file, pickle_save_to_file
+from osbot_utils.utils.Files import path_combine, folder_create, temp_folder_current, file_exists, \
+    pickle_load_from_file, pickle_save_to_file, files_list, files_recursive
 from osbot_utils.utils.Misc import str_md5
 from osbot_utils.utils.Python_Logger import logger_info
 
-FOLDER_CACHE_ROOT_FOLDER = '_cache_pickle'
-SUPPORTED_PARAMS_TYPES   = [int, float, bytearray, bytes, bool, complex, str]
+
 
 class Cache_Pickle:
 
+    _cache__FOLDER_CACHE_ROOT_FOLDER = '_cache_pickle'
+    _cache__SUPPORTED_PARAMS_TYPES   = [int, float, bytearray, bytes, bool, complex, str]
+
     def __init__(self):
-        self.cache_enabled = True
-        self.log_info      = logger_info()
-        self.cache_setup()              # make sure the cache folder exists
+        self._cache_enabled = True
+        #self.log_info      = logger_info()
+        self._cache_setup()              # make sure the cache folder exists
 
+    def __enter__(self): return self
+    def __exit__ (self, type, value, traceback): pass
 
     def __getattribute__(self, name):
-        if name.startswith('cache_') or name.startswith('__'):
-            return super().__getattribute__(name)
-
-        target = super().__getattribute__(name)
-        if not callable(target):
-            return target
-
-        return self.cache_data(target)
-
-    def cache_clear(self):
-        cache_dir = self.cache_path()
+        if name.startswith('_cache_') or name.startswith('__'):                 # if the method is a method from Cache_Pickleor a private method
+            return super().__getattribute__(name)                               # just return it's value
+        target = super().__getattribute__(name)                                 # get the target
+        if not callable(target):                                                # if it is not a function
+            return target                                                       # just return it
+        return self._cache_data(target)                                         # if it is a function, create a wrapper around it
+
+    def _cache_clear(self):
+        cache_dir = self._cache_path()
         for filename in os.listdir(cache_dir):
             if filename.endswith('.pickle'):
                 os.remove(os.path.join(cache_dir, filename))
         return self
 
-    def cache_data(self, func):
+    def _cache_data(self, func):
 
         @wraps(func)
         def wrapper(*args, **kwargs):
@@ -52,60 +55,73 @@ def wrapper(*args, **kwargs):
                 del kwargs['use_cache']
             else:
                 use_cache = True
+
             # after processing these extra params we can resolve the file name and check if it exists
-            cache_file_name = self.cache_resolve_file_name(func, args, kwargs)
+            cache_file_name = self._cache_resolve_file_name(func, args, kwargs)
 
-            # path_file = path_combine(self.cache_path(), f'{caller_name}.pickle')
-            path_file = path_combine(self.cache_path(), cache_file_name)
+            # path_file = path_combine(self._cache_path(), f'{caller_name}.pickle')
+            path_file = path_combine(self._cache_path(), cache_file_name)
 
             if use_cache is True and reload_cache is False and file_exists(path_file):
                 return pickle_load_from_file(path_file)
             else:
                 data = func(*args, **kwargs)
                 if data and use_cache is True:
                     caller_name = func.__name__
-                    self.log_info(f"Saving cache file data for: {caller_name}")
+                    print(f"Saving cache file data for: {caller_name}")
                     pickle_save_to_file(data, path_file)
                 return data
         return wrapper
 
-    def cache_disable(self):
-        self.cache_enabled = False
+    def _cache_disable(self):
+        self._cache_enabled = False
+        return self
 
-    def cache_path(self):
+    def _cache_path(self):
+        class_name  = self.__class__.__name__
         module_name = self.__class__.__module__
-        folder_name = f'{FOLDER_CACHE_ROOT_FOLDER}/{module_name.replace(".", "/")}'
+        folder_name = f'{self._cache__FOLDER_CACHE_ROOT_FOLDER}/{module_name.replace(".", "/")}/{class_name}'
         return path_combine(temp_folder_current(), folder_name)
 
+    def _cache_files(self):
+        return files_recursive(self._cache_path())
 
-    def cache_setup(self):
-        folder_create(self.cache_path())
+    def _cache_setup(self):
+        folder_create(self._cache_path())
         return self
 
-    def cache_kwargs_to_str(self, kwargs):
+    def _cache_kwargs_to_str(self, kwargs):
         kwargs_values_as_str = ''
         if kwargs:
+            if type(kwargs) is not dict:
+                return str(kwargs)
             for key, value in kwargs.items():
-                if type(value) in SUPPORTED_PARAMS_TYPES:
-                    kwargs_values_as_str += f'{key}:{value}|'
+                if value and type(value) not in self._cache__SUPPORTED_PARAMS_TYPES:
+                    value = '(...)'
+                kwargs_values_as_str += f'{key}:{value}|'
         return kwargs_values_as_str
 
-    def cache_args_to_str(self, args):
+    def _cache_args_to_str(self, args):
         args_values_as_str = ''
         if args:
+            if type(args) is not list:
+                return str(args)
             for arg in args:
-                if type(arg) in SUPPORTED_PARAMS_TYPES:
-                    args_values_as_str += str(arg)
+                if not arg or type(arg) in self._cache__SUPPORTED_PARAMS_TYPES:
+                    arg_value = str(arg)
+                else:
+                    arg_value = '(...)'
+                args_values_as_str += f'{arg_value}|'
         return args_values_as_str
 
-    def cache_resolve_file_name(self, function, args=None, kwargs=None):
+    def _cache_resolve_file_name(self, function, args=None, kwargs=None):
         key_name               = function.__name__
         args_md5               = ''
         kwargs_md5             = ''
-        args_values_as_str     = self.cache_args_to_str(args)
-        kwargs_values_as_str   = self.cache_kwargs_to_str(kwargs)
-        if args_values_as_str  : args_md5   = '_' + str_md5(args_values_as_str  )
-        if kwargs_values_as_str: kwargs_md5 = '_' + str_md5(kwargs_values_as_str)
+        args_values_as_str     = self._cache_args_to_str(args)
+        kwargs_values_as_str   = self._cache_kwargs_to_str(kwargs)
+        if args_values_as_str  : args_md5   = '_' + str_md5(args_values_as_str  )[:10]
+        if kwargs_values_as_str: kwargs_md5 = '_' + str_md5(kwargs_values_as_str)[:10]
         cache_file_name        = f'{key_name}{args_md5}{kwargs_md5}'
         cache_file_name       += '.pickle'
         return cache_file_name

osbot_utils/decorators/methods/context.py

@@ -0,0 +1,11 @@
+from contextlib import contextmanager
+
+@contextmanager
+def context(target, exec_before=None, exec_after=None):
+    if exec_before:
+        exec_before()
+    try:
+        yield target
+    finally:
+        if exec_after:
+            exec_after()

tests/base_classes/test_Cache_Pickle.py

@@ -0,0 +1,104 @@
+import types
+from unittest import TestCase
+
+from osbot_utils.base_classes.Cache_Pickle import Cache_Pickle
+from osbot_utils.decorators.methods.context import context
+from osbot_utils.utils.Dev import pprint
+from osbot_utils.utils.Files import folder_exists, current_temp_folder, pickle_load_from_file
+from osbot_utils.utils.Misc import date_time_now, date_now, str_md5
+
+
+class test_Cache_Pickle(TestCase):
+
+    def setUp(self) -> None:
+        self.cache_pickle = Cache_Pickle()
+
+    def test_aaaa(self):
+        pass
+
+    def test__init__(self):
+        with self.cache_pickle as _:
+            assert _._cache_enabled is True
+            assert _._cache_files() == []
+
+        assert Cache_Pickle._cache__FOLDER_CACHE_ROOT_FOLDER ==  '_cache_pickle'
+        assert Cache_Pickle._cache__SUPPORTED_PARAMS_TYPES   == [int, float, bytearray, bytes, bool, complex, str]
+
+
+    def test__cache_data(self):
+        class An_Class(Cache_Pickle):
+            def return_42(self):
+                return 42
+
+
+
+        an_class = An_Class()
+        assert isinstance(an_class , An_Class    )
+        assert isinstance(an_class , Cache_Pickle)
+        assert an_class._cache_path().endswith('_cache_pickle/test_Cache_Pickle/An_Class')
+        assert an_class._cache_clear()   is an_class
+        assert an_class._cache_files()  == []
+        assert An_Class().return_42() == 42
+        cache_files = an_class._cache_files()
+        cache_file  = cache_files[0]
+        assert cache_file.endswith('_cache_pickle/test_Cache_Pickle/An_Class/return_42.pickle')
+        assert len(cache_files) == 1
+        assert pickle_load_from_file(cache_file)
+
+
+
+    def test__cache_disable(self):
+        with self.cache_pickle as _:
+            assert _._cache_disable() == _
+            self._cache_enabled = False
+        return self
+
+    def test__cache_path(self):
+        cache_path = self.cache_pickle._cache_path()
+        assert folder_exists(cache_path)
+        assert cache_path.endswith  ("_cache_pickle/osbot_utils/base_classes/Cache_Pickle")
+        assert cache_path.startswith(current_temp_folder())
+
+    def test__cache_kwargs_to_str(self):
+        with context(self.cache_pickle._cache_kwargs_to_str) as _:
+            assert _({}                   ) == ''
+            assert _({'a':1}              ) == 'a:1|'
+            assert _({'a':1,'b':2}        ) == 'a:1|b:2|'
+            assert _({'a':1,'b':2,'c':3  }) == 'a:1|b:2|c:3|'
+            assert _({'aaaaa':'bbbb'     }) == 'aaaaa:bbbb|'
+            assert _({'tttt':Cache_Pickle}) == 'tttt:(...)|'            # for the values not in _cache__SUPPORTED_PARAMS_TYPES
+            assert _({'nnnn': None       }) == 'nnnn:None|'
+            assert _({None: None         }) == 'None:None|'
+            assert _({None}               ) == '{None}'
+            assert _(''                   ) == ''
+            assert _(None                 ) == ''
+            assert _(0                    ) == ''
+            assert _(1                    ) == '1'
+            assert _(str                  ) == "<class 'str'>"
+
+    def test__cache_args_to_str(self):
+        with context(self.cache_pickle._cache_args_to_str) as _:
+            assert _([]               ) == ''
+            assert _([1]              ) == '1|'
+            assert _([1,'2']          ) == '1|2|'
+            assert _([None]           ) == 'None|'
+            assert _([1, None]        ) == '1|None|'
+            assert _([1, Cache_Pickle]) == '1|(...)|'                   # for the values not in _cache__SUPPORTED_PARAMS_TYPES
+            assert _(None             ) == ''
+            assert _(0                ) == ''
+            assert _(''               ) == ''
+            assert _(1                ) == '1'
+            assert _(str              ) == "<class 'str'>"
+
+    def test___cache_resolve_file_name(self):
+        def an_function(): pass
+
+        with context(self.cache_pickle._cache_resolve_file_name) as _:
+            assert _(an_function              ) == 'an_function.pickle'
+            assert _(an_function, None, None  ) == 'an_function.pickle'
+            assert _(an_function, []  , None  ) == 'an_function.pickle'
+            assert _(an_function, []  , {}    ) == 'an_function.pickle'
+            assert _(an_function, [0]         ) == 'an_function_8b879ac2fa.pickle'
+            assert _(an_function, [1]         ) == 'an_function_77529b68d1.pickle'
+            assert _(an_function, [] , {'a':1}) == 'an_function_e280cf8acd.pickle'
+            assert _(an_function, [1], {'a':1}) == 'an_function_77529b68d1_e280cf8acd.pickle'

tests/decorators/methods/test_context.py

@@ -0,0 +1,63 @@
+from unittest import TestCase
+from unittest.mock import MagicMock
+
+from osbot_utils.decorators.methods.context import context
+
+
+class test_context(TestCase):
+
+    def test_context(self):
+        before_mock = MagicMock()
+        after_mock = MagicMock()
+        def the_answer():
+            return 42
+        def exec_before():
+            before_mock()
+        def exec_after():
+            after_mock()
+
+        with context(the_answer, exec_before=exec_before, exec_after=exec_after) as target:
+            assert target() == 42
+
+        before_mock.assert_called_once()
+        after_mock.assert_called_once()
+
+    def test_exec_before_and_after_called(self):
+        before_mock = MagicMock()
+        after_mock = MagicMock()
+        target_value = "test_target"
+
+        with context(target_value, exec_before=before_mock, exec_after=after_mock) as target:
+            before_mock.assert_called_once()
+            after_mock.assert_not_called()
+            self.assertEqual(target, target_value)
+
+        after_mock.assert_called_once()
+
+    def test_confirm_yield_is_target(self):
+        target_value = "test_target"
+        with context(target_value) as target:               # No before or after functions provided, should run without errors
+            self.assertEqual(target, target_value)
+
+
+    def test_exec_after_called_even_if_exception_raised(self):
+        after_mock = MagicMock()
+
+        with self.assertRaises(ValueError):
+            with context("test_target", exec_after=after_mock) as target:
+                raise ValueError("An error occurred inside the with block.")
+
+        after_mock.assert_called_once()
+
+        after_mock_2 = MagicMock()
+        def throw_exception():
+            raise ValueError("An error occurred inside the with block.")
+
+        with self.assertRaises(ValueError):
+            with context(throw_exception, exec_after=after_mock_2) as target:
+                target()
+
+        after_mock_2.assert_called_once()
+
+
+