Added CacheInterface abstraction

Author: vladvildanov

redis/cache.py

@@ -1,5 +1,6 @@
 from abc import ABC, abstractmethod
 from enum import Enum
+from typing import Any, Hashable
 
 from cachetools import Cache, LFUCache, LRUCache, RRCache, TTLCache
 
@@ -121,31 +122,112 @@ def is_allowed_to_cache(self, command: str) -> bool:
         return command in self.DEFAULT_ALLOW_LIST
 
 
-class CacheClass(Enum):
+class EvictionPolicyCacheClass(Enum):
     LRU = LRUCache
     LFU = LFUCache
     RANDOM = RRCache
     TTL = TTLCache
 
 
+class CacheClassEvictionPolicy(Enum):
+    LRUCache = EvictionPolicy.LRU
+    LFUCache = EvictionPolicy.LFU
+    RRCache = EvictionPolicy.RANDOM
+    TTLCache = EvictionPolicy.TTL
+
+
+class CacheInterface(ABC):
+
+    @property
+    @abstractmethod
+    def currsize(self) -> float:
+        pass
+
+    @property
+    @abstractmethod
+    def maxsize(self) -> float:
+        pass
+
+    @property
+    @abstractmethod
+    def eviction_policy(self) -> EvictionPolicy:
+        pass
+
+    @abstractmethod
+    def get(self, key: Hashable, default: Any = None):
+        pass
+
+    @abstractmethod
+    def set(self, key: Hashable, value: Any):
+        pass
+
+    @abstractmethod
+    def exists(self, key: Hashable) -> bool:
+        pass
+
+    @abstractmethod
+    def remove(self, key: Hashable):
+        pass
+
+    @abstractmethod
+    def clear(self):
+        pass
+
+
 class CacheFactoryInterface(ABC):
     @abstractmethod
-    def get_cache(self) -> Cache:
+    def get_cache(self) -> CacheInterface:
         pass
 
 
-class CacheFactory(CacheFactoryInterface):
+class CacheToolsFactory(CacheFactoryInterface):
     def __init__(self, conf: CacheConfiguration):
         self._conf = conf
 
-    def get_cache(self) -> Cache:
+    def get_cache(self) -> CacheInterface:
         eviction_policy = self._conf.get_eviction_policy()
         cache_class = self._get_cache_class(eviction_policy).value
 
         if eviction_policy == EvictionPolicy.TTL:
-            return cache_class(self._conf.get_max_size(), self._conf.get_ttl())
+            cache_inst = cache_class(self._conf.get_max_size(), self._conf.get_ttl())
+        else:
+            cache_inst = cache_class(self._conf.get_max_size())
+
+        return CacheToolsAdapter(cache_inst)
+
+    def _get_cache_class(
+        self, eviction_policy: EvictionPolicy
+    ) -> EvictionPolicyCacheClass:
+        return EvictionPolicyCacheClass[eviction_policy.value]
+
+
+class CacheToolsAdapter(CacheInterface):
+    def __init__(self, cache: Cache):
+        self._cache = cache
+
+    def get(self, key: Hashable, default: Any = None):
+        return self._cache.get(key, default)
+
+    def set(self, key: Hashable, value: Any):
+        self._cache[key] = value
+
+    def exists(self, key: Hashable) -> bool:
+        return key in self._cache
+
+    def remove(self, key: Hashable):
+        self._cache.pop(key)
+
+    def clear(self):
+        self._cache.clear()
+
+    @property
+    def currsize(self) -> float:
+        return self._cache.currsize
 
-        return cache_class(self._conf.get_max_size())
+    @property
+    def maxsize(self) -> float:
+        return self._cache.maxsize
 
-    def _get_cache_class(self, eviction_policy: EvictionPolicy) -> CacheClass:
-        return CacheClass[eviction_policy.value]
+    @property
+    def eviction_policy(self) -> EvictionPolicy:
+        return CacheClassEvictionPolicy[self._cache.__class__.__name__].value

redis/client.py

@@ -14,7 +14,7 @@
     _RedisCallbacksRESP3,
     bool_ok,
 )
-from redis.cache import EvictionPolicy
+from redis.cache import CacheInterface, EvictionPolicy
 from redis.commands import (
     CoreCommands,
     RedisModuleCommands,
@@ -215,7 +215,7 @@ def __init__(
         credential_provider: Optional[CredentialProvider] = None,
         protocol: Optional[int] = 2,
         use_cache: bool = False,
-        cache: Optional[Cache] = None,
+        cache: Optional[CacheInterface] = None,
         cache_eviction: Optional[EvictionPolicy] = None,
         cache_size: int = 128,
         cache_ttl: int = 300,
@@ -603,7 +603,7 @@ def parse_response(self, connection, command_name, **options):
             return self.response_callbacks[command_name](response, **options)
         return response
 
-    def get_cache(self) -> Optional[Cache]:
+    def get_cache(self) -> Optional[CacheInterface]:
         return self.connection_pool.cache
 
 

redis/cluster.py

@@ -10,7 +10,7 @@
 from redis._parsers import CommandsParser, Encoder
 from redis._parsers.helpers import parse_scan
 from redis.backoff import default_backoff
-from redis.cache import EvictionPolicy
+from redis.cache import CacheInterface, EvictionPolicy
 from redis.client import CaseInsensitiveDict, PubSub, Redis
 from redis.commands import READ_COMMANDS, RedisClusterCommands
 from redis.commands.helpers import list_or_args
@@ -170,6 +170,9 @@ def parse_cluster_myshardid(resp, **options):
     "unix_socket_path",
     "username",
     "use_cache",
+    "cache",
+    "cache_size",
+    "cache_ttl",
 )
 KWARGS_DISABLED_KEYS = ("host", "port")
 
@@ -504,7 +507,7 @@ def __init__(
         url: Optional[str] = None,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
         use_cache: bool = False,
-        cache: Optional[Cache] = None,
+        cache: Optional[CacheInterface] = None,
         cache_eviction: Optional[EvictionPolicy] = None,
         cache_size: int = 128,
         cache_ttl: int = 300,

redis/connection.py

@@ -15,7 +15,12 @@
 from apscheduler.schedulers.background import BackgroundScheduler
 from cachetools import Cache, LRUCache
 from cachetools.keys import hashkey
-from redis.cache import CacheConfiguration, CacheFactory
+from redis.cache import (
+    CacheConfiguration,
+    CacheFactoryInterface,
+    CacheInterface,
+    CacheToolsFactory,
+)
 
 from ._parsers import Encoder, _HiredisParser, _RESP2Parser, _RESP3Parser
 from .backoff import NoBackoff
@@ -728,7 +733,7 @@ class CacheProxyConnection(ConnectionInterface):
     def __init__(
         self,
         conn: ConnectionInterface,
-        cache: Cache,
+        cache: CacheInterface,
         conf: CacheConfiguration,
         cache_lock: threading.Lock,
     ):
@@ -806,7 +811,7 @@ def send_command(self, *args, **kwargs):
 
             # Set temporary entry as a status to prevent
             # race condition from another connection.
-            self._cache[self._current_command_hash] = "caching-in-progress"
+            self._cache.set(self._current_command_hash, "caching-in-progress")
 
         # Send command over socket only if it's allowed
         # read-only command that not yet cached.
@@ -821,10 +826,10 @@ def read_response(
         with self._cache_lock:
             # Check if command response exists in a cache and it's not in progress.
             if (
-                self._current_command_hash in self._cache
-                and self._cache[self._current_command_hash] != "caching-in-progress"
+                self._cache.exists(self._current_command_hash)
+                and self._cache.get(self._current_command_hash) != "caching-in-progress"
             ):
-                return copy.deepcopy(self._cache[self._current_command_hash])
+                return copy.deepcopy(self._cache.get(self._current_command_hash))
 
         response = self._conn.read_response(
             disable_decoding=disable_decoding,
@@ -835,7 +840,7 @@ def read_response(
         with self._cache_lock:
             # If response is None prevent from caching.
             if response is None:
-                self._cache.pop(self._current_command_hash)
+                self._cache.remove(self._current_command_hash)
                 return response
             # Prevent not-allowed command from caching.
             elif self._current_command_hash is None:
@@ -855,7 +860,7 @@ def read_response(
             # Cache only responses that still valid
             # and wasn't invalidated by another connection in meantime.
             if cache_entry is not None:
-                self._cache[self._current_command_hash] = response
+                self._cache.set(self._current_command_hash, response)
 
         return response
 
@@ -892,7 +897,7 @@ def _on_invalidation_callback(self, data: List[Union[str, Optional[List[str]]]])
                         # Make sure that all command responses
                         # associated with this key will be deleted
                         for cache_key in self._keys_mapping[normalized_key]:
-                            self._cache.pop(cache_key)
+                            self._cache.remove(cache_key)
                     # Removes key from mapping cache
                     self._keys_mapping.pop(normalized_key)
 
@@ -1246,6 +1251,7 @@ def __init__(
         self,
         connection_class=Connection,
         max_connections: Optional[int] = None,
+        cache_factory: Optional[CacheFactoryInterface] = None,
         **connection_kwargs,
     ):
         max_connections = max_connections or 2**31
@@ -1257,7 +1263,7 @@ def __init__(
         self.max_connections = max_connections
         self.cache = None
         self._cache_conf = None
-        self._cache_factory = None
+        self._cache_factory = cache_factory
         self.cache_lock = None
         self.scheduler = None
 
@@ -1269,11 +1275,17 @@ def __init__(
             self._cache_lock = threading.Lock()
 
             cache = self.connection_kwargs.get("cache")
+
             if cache is not None:
+                if not isinstance(cache, CacheInterface):
+                    raise ValueError("Cache must implement CacheInterface")
+
                 self.cache = cache
             else:
-                cache_factory = CacheFactory(self._cache_conf)
-                self.cache = cache_factory.get_cache()
+                if self._cache_factory is not None:
+                    self.cache = self._cache_factory.get_cache()
+                else:
+                    self.cache = CacheToolsFactory(self._cache_conf).get_cache()
 
             self.scheduler = BackgroundScheduler()
             self.scheduler.add_job(