ignore caching errors

smart_imports/cache.py

@@ -1,10 +1,16 @@
 
 import os
+import pathlib
 import hashlib
+import warnings
+import functools
 
 from . import constants
 
 
+WARNING_MESSAGE = 'Error while accessing smart imports cache'
+
+
 def get_checksum(source):
     return hashlib.sha256(source.encode('utf-8')).hexdigest()
 
@@ -13,6 +19,20 @@ def get_cache_path(cache_dir, module_name):
     return os.path.join(cache_dir, module_name + '.cache')
 
 
+def ignore_errors(function):
+
+    @functools.wraps(function)
+    def wrapper(*argv, **kwargs):
+        try:
+            return function(*argv, **kwargs)
+        except Exception as e:
+            warnings.warn('{}: {}'.format(WARNING_MESSAGE, e), UserWarning, stacklevel=2)
+            return None
+
+    return wrapper
+
+
+@ignore_errors
 def get(cache_dir, module_name, checksum):
 
     cache_path = get_cache_path(cache_dir, module_name)
@@ -36,7 +56,10 @@ def get(cache_dir, module_name, checksum):
     return variables
 
 
+@ignore_errors
 def set(cache_dir, module_name, checksum, variables):
+    pathlib.Path(cache_dir).mkdir(parents=True, exist_ok=True)
+
     cache_path = get_cache_path(cache_dir, module_name)
 
     with open(cache_path, 'w') as f:

smart_imports/tests/test_cache.py

@@ -3,6 +3,7 @@
 import uuid
 import unittest
 import tempfile
+import warnings
 
 from unittest import mock
 
@@ -54,6 +55,56 @@ def test_set_get(self):
 
             self.assertEqual(variables, loaded_variables)
 
+    def test_set_get__create_directories(self):
+        variables = ['a', 'x', 'zzz', 'long_long_long']
+
+        with tempfile.TemporaryDirectory() as temp_directory:
+            cache_dir = os.path.join(temp_directory, 'unexisted_1', 'unexisted_2')
+
+            cache.set(cache_dir=cache_dir,
+                      module_name='x.y',
+                      checksum=cache.get_checksum('abc'),
+                      variables=variables)
+
+            self.assertTrue(os.path.isdir(cache_dir))
+
+            self.assertTrue(os.path.isfile(os.path.join(cache_dir, 'x.y.cache')))
+
+            loaded_variables = cache.get(cache_dir=cache_dir,
+                                         module_name='x.y',
+                                         checksum=cache.get_checksum('abc'))
+
+            self.assertEqual(variables, loaded_variables)
+
+    def test_set_get__ignore_errors(self):
+        variables = ['a', 'x', 'zzz', 'long_long_long']
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+
+            cache.set(cache_dir=None,
+                      module_name='x.y',
+                      checksum=cache.get_checksum('abc'),
+                      variables=variables)
+
+            assert len(w) == 1
+            assert issubclass(w[-1].category, UserWarning)
+            assert cache.WARNING_MESSAGE in str(w[-1].message)
+
+        with warnings.catch_warnings(record=True) as w:
+            warnings.simplefilter("always")
+
+            loaded_variables = cache.get(cache_dir=None,
+                                         module_name='x.y',
+                                         checksum=cache.get_checksum('abc'))
+
+            assert len(w) == 1
+            assert issubclass(w[-1].category, UserWarning)
+            assert cache.WARNING_MESSAGE in str(w[-1].message)
+
+            self.assertEqual(loaded_variables, None)
+
+
     def test_wrong_checksum(self):
         with tempfile.TemporaryDirectory() as temp_directory:
             cache.set(cache_dir=temp_directory,