refactor: use single method for error_handling in CustomDjangoCache

kolibri/deployment/default/custom_django_cache.py

@@ -15,29 +15,27 @@ class CustomDjangoCache(DjangoCache):
 
     ERRORS_TO_HANDLE = (sqlite3.OperationalError, AssertionError)
 
-    def try_execute_return_none_on_error(self, method, *args, **kwargs):
-        try:
-            return method(*args, **kwargs)
-        except self.ERRORS_TO_HANDLE:
-            return None
-
-    def try_execute_return_false_on_error(self, method, *args, **kwargs):
-        try:
-            return method(*args, **kwargs)
-        except self.ERRORS_TO_HANDLE:
-            return False
-
-    def try_execute_no_return_on_error(self, method, *args, **kwargs):
-        try:
-            method(*args, **kwargs)
-        except self.ERRORS_TO_HANDLE:
-            pass
-
-    def try_execute_return_zero_on_error(self, method, *args, **kwargs):
+    def try_execute(self, method_name, error_return_value, *args, **kwargs):
+        """
+        Safely executes a method with error handling.
+            :param method_name --> (str): name of method to execute
+            :param error_return_value --> (any): value to return if error occur
+            :param *args: positional arguments for method
+            :param *kwargs: keyword arguments for method
+            :return: The return value of the executed method if successful,
+                     otherwise returns error_return_value
+        """
         try:
+            method = getattr(super(CustomDjangoCache, self), method_name)
+            if method is None:
+                raise ValueError(
+                    "{method_name} is not a valid method".format(
+                        method_name=method_name
+                    )
+                )
             return method(*args, **kwargs)
         except self.ERRORS_TO_HANDLE:
-            return 0
+            return error_return_value
 
     def add(
         self,
@@ -49,15 +47,16 @@ def add(
         tag=None,
         retry=True,
     ):
-        return self.try_execute_return_false_on_error(
-            super(CustomDjangoCache, self).add,
-            key,
-            value,
-            timeout,
-            version,
-            read,
-            tag,
-            retry,
+        return self.try_execute(
+            method_name="add",
+            error_return_value=False,
+            key=key,
+            value=value,
+            timeout=timeout,
+            version=version,
+            read=read,
+            tag=tag,
+            retry=retry,
         )
 
     def has_key(self, key, version=None):
@@ -68,8 +67,8 @@ def has_key(self, key, version=None):
         :return: True if key is found
 
         """
-        return self.try_execute_return_false_on_error(
-            super(CustomDjangoCache, self).has_key, key, version=version
+        return self.try_execute(
+            method_name="has_key", error_return_value=False, key=key, version=version
         )
 
     def get(
@@ -82,15 +81,16 @@ def get(
         tag=False,
         retry=False,
     ):
-        return self.try_execute_return_none_on_error(
-            super(CustomDjangoCache, self).get,
-            key,
-            default,
-            version,
-            read,
-            expire_time,
-            tag,
-            retry,
+        return self.try_execute(
+            method_name="get",
+            error_return_value=None,
+            key=key,
+            default=default,
+            version=version,
+            read=read,
+            expire_time=expire_time,
+            tag=tag,
+            retry=retry,
         )
 
     def set(
@@ -103,77 +103,95 @@ def set(
         tag=None,
         retry=True,
     ):
-        return self.try_execute_return_false_on_error(
-            super(CustomDjangoCache, self).set,
-            key,
-            value,
-            timeout,
-            version,
-            read,
-            tag,
-            retry,
+        return self.try_execute(
+            method_name="set",
+            error_return_value=False,
+            key=key,
+            value=value,
+            timeout=timeout,
+            version=version,
+            read=read,
+            tag=tag,
+            retry=retry,
         )
 
     def touch(self, key, timeout=DEFAULT_TIMEOUT, version=None, retry=True):
-        return self.try_execute_return_false_on_error(
-            super(CustomDjangoCache, self).touch, key, timeout, version, retry
+        return self.try_execute(
+            method_name="touch",
+            error_return_value=False,
+            key=key,
+            timeout=timeout,
+            version=version,
+            retry=retry,
         )
 
     def pop(
         self, key, default=None, version=None, expire_time=False, tag=False, retry=True
     ):
-        return self.try_execute_return_none_on_error(
-            super(CustomDjangoCache, self).pop,
-            key,
-            default,
-            version,
-            expire_time,
-            tag,
-            retry,
+        return self.try_execute(
+            method_name="pop",
+            error_return_value=None,
+            key=key,
+            default=default,
+            version=version,
+            expire_time=expire_time,
+            tag=tag,
+            retry=retry,
         )
 
     def delete(self, key, version=None, retry=True):
-        self.try_execute_no_return_on_error(
-            super(CustomDjangoCache, self).delete, key, version, retry
+        self.try_execute(
+            method_name="delete",
+            error_return_value=None,
+            key=key,
+            version=version,
+            retry=retry,
         )
 
     def incr(self, key, delta=1, version=None, default=None, retry=True):
-        return self.try_execute_return_none_on_error(
-            super(CustomDjangoCache, self).incr, key, delta, version, default, retry
+        return self.try_execute(
+            method_name="incr",
+            error_return_value=None,
+            key=key,
+            delta=delta,
+            version=version,
+            default=default,
+            retry=retry,
         )
 
     def decr(self, key, delta=1, version=None, default=None, retry=True):
-        return self.try_execute_return_none_on_error(
-            super(CustomDjangoCache, self).decr, key, delta, version, default, retry
+        return self.try_execute(
+            method_name="decr",
+            key=key,
+            error_return_value=None,
+            delta=delta,
+            version=version,
+            default=default,
+            retry=retry,
         )
 
     def expire(self, retry=False):
-        return self.try_execute_return_zero_on_error(super().expire, retry)
+        return self.try_execute(method_name="expire", error_return_value=0, retry=retry)
 
     def stats(self, enable=True, reset=False):
-        result = self.try_execute_return_none_on_error(super().stats, enable, reset)
-        return result if isinstance(result, tuple) else (0, 0)
+        return self.try_execute(
+            method_name="stats", error_return_value=(0, 0), enable=enable, reset=reset
+        )
 
     def create_tag_index(self):
-        self.try_execute_no_return_on_error(
-            super(CustomDjangoCache, self).create_tag_index
-        )
+        return self.try_execute(method_name="create_tag_index", error_return_value=None)
 
     def drop_tag_index(self):
-        self.try_execute_no_return_on_error(
-            super(CustomDjangoCache, self).drop_tag_index
-        )
+        return self.try_execute(method_name="drop_tag_index", error_return_value=None)
 
     def evict(self, tag):
-        return self.try_execute_return_zero_on_error(super().evict, tag)
+        return self.try_execute(method_name="evict", error_return_value=0, tag=tag)
 
     def cull(self):
-        return self.try_execute_return_zero_on_error(super().cull)
+        return self.try_execute(method_name="cull", error_return_value=0)
 
     def clear(self):
-        return self.try_execute_return_zero_on_error(super().clear)
+        return self.try_execute(method_name="clear", error_return_value=0)
 
     def close(self, **kwargs):
-        self.try_execute_no_return_on_error(
-            super(CustomDjangoCache, self).close, **kwargs
-        )
+        return self.try_execute(method_name="close", error_return_value=None, **kwargs)