rccache: refactor and update code to support latest dogpile code changes (mostly on custom serializers)

rhodecode/lib/rc_cache/__init__.py

@@ -20,21 +20,22 b''
 
 import logging
 from dogpile.cache import register_backend
+module_name = 'rhodecode'
 
 register_backend(
-    "dogpile.cache.rc.memory_lru", "rhodecode.lib.rc_cache.backends",
+    "dogpile.cache.rc.memory_lru", f"{module_name}.lib.rc_cache.backends",
     "LRUMemoryBackend")
 
 register_backend(
-    "dogpile.cache.rc.file_namespace", "rhodecode.lib.rc_cache.backends",
+    "dogpile.cache.rc.file_namespace", f"{module_name}.lib.rc_cache.backends",
     "FileNamespaceBackend")
 
 register_backend(
-    "dogpile.cache.rc.redis", "rhodecode.lib.rc_cache.backends",
+    "dogpile.cache.rc.redis", f"{module_name}.lib.rc_cache.backends",
     "RedisPickleBackend")
 
 register_backend(
-    "dogpile.cache.rc.redis_msgpack", "rhodecode.lib.rc_cache.backends",
+    "dogpile.cache.rc.redis_msgpack", f"{module_name}.lib.rc_cache.backends",
     "RedisMsgPackBackend")
 
 

rhodecode/lib/rc_cache/backends.py

@@ -21,22 +21,27 b''
 import time
 import errno
 import logging
+import functools
 
 import msgpack
 import redis
 import gevent
+import pickle
+import fcntl
+flock_org = fcntl.flock
+from typing import Union
 
-from dogpile.cache.api import CachedValue
 from dogpile.cache.backends import memory as memory_backend
 from dogpile.cache.backends import file as file_backend
 from dogpile.cache.backends import redis as redis_backend
-from dogpile.cache.backends.file import NO_VALUE, FileLock
+from dogpile.cache.backends.file import FileLock
 from dogpile.cache.util import memoized_property
+from dogpile.cache.api import Serializer, Deserializer
 
 from pyramid.settings import asbool
 
 from rhodecode.lib.memory_lru_dict import LRUDict, LRUDictDebug
-from rhodecode.lib.utils import safe_str
+from rhodecode.lib.str_utils import safe_str
 
 
 _default_max_size = 1024
@@ -70,55 +75,22 b' class LRUMemoryBackend(memory_backend.Me'
             self.delete(key)
 
 
-class PickleSerializer(object):
-
-    def _dumps(self, value, safe=False):
-        try:
-            return pickle.dumps(value)
-        except Exception:
-            if safe:
-                return NO_VALUE
-            else:
-                raise
-
-    def _loads(self, value, safe=True):
-        try:
-            return pickle.loads(value)
-        except Exception:
-            if safe:
-                return NO_VALUE
-            else:
-                raise
+class PickleSerializer:
+    serializer: Union[None, Serializer] = staticmethod(  # type: ignore
+        functools.partial(pickle.dumps, protocol=pickle.HIGHEST_PROTOCOL)
+    )
+    deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
+        functools.partial(pickle.loads)
+    )
 
 
 class MsgPackSerializer(object):
-
-    def _dumps(self, value, safe=False):
-        try:
-            return msgpack.packb(value)
-        except Exception:
-            if safe:
-                return NO_VALUE
-            else:
-                raise
-
-    def _loads(self, value, safe=True):
-        """
-        pickle maintained the `CachedValue` wrapper of the tuple
-        msgpack does not, so it must be added back in.
-       """
-        try:
-            value = msgpack.unpackb(value, use_list=False)
-            return CachedValue(*value)
-        except Exception:
-            if safe:
-                return NO_VALUE
-            else:
-                raise
-
-
-import fcntl
-flock_org = fcntl.flock
+    serializer: Union[None, Serializer] = staticmethod(  # type: ignore
+        msgpack.packb
+    )
+    deserializer: Union[None, Deserializer] = staticmethod(  # type: ignore
+        functools.partial(msgpack.unpackb, use_list=False)
+    )
 
 
 class CustomLockFactory(FileLock):
@@ -195,36 +167,6 b' class FileNamespaceBackend(PickleSeriali'
     def get_store(self):
         return self.filename
 
-    def _dbm_get(self, key):
-        with self._dbm_file(False) as dbm:
-            if hasattr(dbm, 'get'):
-                value = dbm.get(key, NO_VALUE)
-            else:
-                # gdbm objects lack a .get method
-                try:
-                    value = dbm[key]
-                except KeyError:
-                    value = NO_VALUE
-            if value is not NO_VALUE:
-                value = self._loads(value)
-            return value
-
-    def get(self, key):
-        try:
-            return self._dbm_get(key)
-        except Exception:
-            log.error('Failed to fetch DBM key %s from DB: %s', key, self.get_store())
-            raise
-
-    def set(self, key, value):
-        with self._dbm_file(True) as dbm:
-            dbm[key] = self._dumps(value)
-
-    def set_multi(self, mapping):
-        with self._dbm_file(True) as dbm:
-            for key, value in mapping.items():
-                dbm[key] = self._dumps(value)
-
 
 class BaseRedisBackend(redis_backend.RedisBackend):
     key_prefix = ''
@@ -251,58 +193,26 b' class BaseRedisBackend(redis_backend.Red'
             )
 
         connection_pool = redis.ConnectionPool(**args)
-
-        return redis.StrictRedis(connection_pool=connection_pool)
+        self.writer_client = redis.StrictRedis(
+            connection_pool=connection_pool
+        )
+        self.reader_client = self.writer_client
 
     def list_keys(self, prefix=''):
         prefix = '{}:{}*'.format(self.key_prefix, prefix)
-        return self.client.keys(prefix)
+        return self.reader_client.keys(prefix)
 
     def get_store(self):
-        return self.client.connection_pool
-
-    def get(self, key):
-        value = self.client.get(key)
-        if value is None:
-            return NO_VALUE
-        return self._loads(value)
-
-    def get_multi(self, keys):
-        if not keys:
-            return []
-        values = self.client.mget(keys)
-        loads = self._loads
-        return [
-            loads(v) if v is not None else NO_VALUE
-            for v in values]
-
-    def set(self, key, value):
-        if self.redis_expiration_time:
-            self.client.setex(key, self.redis_expiration_time,
-                              self._dumps(value))
-        else:
-            self.client.set(key, self._dumps(value))
-
-    def set_multi(self, mapping):
-        dumps = self._dumps
-        mapping = dict(
-            (k, dumps(v))
-            for k, v in mapping.items()
-        )
-
-        if not self.redis_expiration_time:
-            self.client.mset(mapping)
-        else:
-            pipe = self.client.pipeline()
-            for key, value in mapping.items():
-                pipe.setex(key, self.redis_expiration_time, value)
-            pipe.execute()
+        return self.reader_client.connection_pool
 
     def get_mutex(self, key):
         if self.distributed_lock:
             lock_key = '_lock_{0}'.format(safe_str(key))
-            return get_mutex_lock(self.client, lock_key, self._lock_timeout,
-                                  auto_renewal=self._lock_auto_renewal)
+            return get_mutex_lock(
+                self.writer_client, lock_key,
+                self._lock_timeout,
+                auto_renewal=self._lock_auto_renewal
+            )
         else:
             return None
 
@@ -318,7 +228,7 b' class RedisMsgPackBackend(MsgPackSeriali'
 
 
 def get_mutex_lock(client, lock_key, lock_timeout, auto_renewal=False):
-    import redis_lock
+    from rhodecode.lib._vendor import redis_lock
 
     class _RedisLockWrapper(object):
         """LockWrapper for redis_lock"""