Removed use_cache, make health_check configurable, removed retry logic around can_read()

Author: vladvildanov

redis/cache.py

@@ -39,6 +39,12 @@ def __init__(
         self.cache_value = cache_value
         self.status = status
 
+    def __hash__(self):
+        return hash((self.cache_key, self.cache_value, self.status))
+
+    def __eq__(self, other):
+        return hash(self) == hash(other)
+
 
 class EvictionPolicyInterface(ABC):
     @property
@@ -68,63 +74,71 @@ def touch(self, cache_key: CacheKey) -> None:
         pass
 
 
-class CacheInterface(ABC):
+class CacheConfigurationInterface(ABC):
     @abstractmethod
-    def get_collection(self) -> OrderedDict:
+    def get_cache_class(self):
         pass
 
     @abstractmethod
-    def get_eviction_policy(self) -> EvictionPolicyInterface:
+    def get_max_size(self) -> int:
         pass
 
     @abstractmethod
-    def get_max_size(self) -> int:
+    def get_eviction_policy(self):
         pass
 
     @abstractmethod
-    def get(self, key: CacheKey) -> Union[CacheEntry, None]:
+    def get_health_check_interval(self) -> float:
         pass
 
     @abstractmethod
-    def set(self, entry: CacheEntry) -> bool:
+    def is_exceeds_max_size(self, count: int) -> bool:
         pass
 
     @abstractmethod
-    def delete_by_cache_keys(self, cache_keys: List[CacheKey]) -> List[bool]:
+    def is_allowed_to_cache(self, command: str) -> bool:
         pass
 
+
+class CacheInterface(ABC):
     @abstractmethod
-    def delete_by_redis_keys(self, redis_keys: List[bytes]) -> List[bool]:
+    def get_collection(self) -> OrderedDict:
         pass
 
     @abstractmethod
-    def flush(self) -> int:
+    def get_config(self) -> CacheConfigurationInterface:
         pass
 
     @abstractmethod
-    def is_cachable(self, key: CacheKey) -> bool:
+    def get_eviction_policy(self) -> EvictionPolicyInterface:
         pass
 
+    @abstractmethod
+    def get_size(self) -> int:
+        pass
 
-class CacheConfigurationInterface(ABC):
     @abstractmethod
-    def get_cache_class(self):
+    def get(self, key: CacheKey) -> Union[CacheEntry, None]:
         pass
 
     @abstractmethod
-    def get_max_size(self) -> int:
+    def set(self, entry: CacheEntry) -> bool:
         pass
 
     @abstractmethod
-    def get_eviction_policy(self):
+    def delete_by_cache_keys(self, cache_keys: List[CacheKey]) -> List[bool]:
         pass
 
     @abstractmethod
-    def is_exceeds_max_size(self, count: int) -> bool:
+    def delete_by_redis_keys(self, redis_keys: List[bytes]) -> List[bool]:
         pass
 
     @abstractmethod
-    def is_allowed_to_cache(self, command: str) -> bool:
+    def flush(self) -> int:
+        pass
+
+    @abstractmethod
+    def is_cachable(self, key: CacheKey) -> bool:
         pass
 
 
@@ -141,11 +155,14 @@ def __init__(
     def get_collection(self) -> OrderedDict:
         return self._cache
 
+    def get_config(self) -> CacheConfigurationInterface:
+        return self._cache_config
+
     def get_eviction_policy(self) -> EvictionPolicyInterface:
         return self._eviction_policy
 
-    def get_max_size(self) -> int:
-        return self._cache_config.get_max_size()
+    def get_size(self) -> int:
+        return len(self._cache)
 
     def set(self, entry: CacheEntry) -> bool:
         if not self.is_cachable(entry.cache_key):
@@ -256,7 +273,7 @@ class EvictionPolicy(Enum):
     LRU = LRUPolicy
 
 
-class CacheConfiguration(CacheConfigurationInterface):
+class CacheConfig(CacheConfigurationInterface):
     DEFAULT_CACHE_CLASS = DefaultCache
     DEFAULT_EVICTION_POLICY = EvictionPolicy.LRU
     DEFAULT_MAX_SIZE = 10000
@@ -343,10 +360,12 @@ def __init__(
         max_size: int = DEFAULT_MAX_SIZE,
         cache_class: Any = DEFAULT_CACHE_CLASS,
         eviction_policy: EvictionPolicy = DEFAULT_EVICTION_POLICY,
+        health_check_interval: float = 2.0,
     ):
         self._cache_class = cache_class
         self._max_size = max_size
         self._eviction_policy = eviction_policy
+        self._health_check_interval = health_check_interval
 
     def get_cache_class(self):
         return self._cache_class
@@ -357,6 +376,9 @@ def get_max_size(self) -> int:
     def get_eviction_policy(self) -> EvictionPolicy:
         return self._eviction_policy
 
+    def get_health_check_interval(self) -> float:
+        return self._health_check_interval
+
     def is_exceeds_max_size(self, count: int) -> bool:
         return count > self._max_size
 
@@ -371,11 +393,11 @@ def get_cache(self) -> CacheInterface:
 
 
 class CacheFactory(CacheFactoryInterface):
-    def __init__(self, cache_config: Optional[CacheConfiguration] = None):
+    def __init__(self, cache_config: Optional[CacheConfig] = None):
         self._config = cache_config
 
         if self._config is None:
-            self._config = CacheConfiguration()
+            self._config = CacheConfig()
 
     def get_cache(self) -> CacheInterface:
         cache_class = self._config.get_cache_class()

redis/client.py

@@ -13,7 +13,7 @@
     _RedisCallbacksRESP3,
     bool_ok,
 )
-from redis.cache import CacheConfiguration, CacheInterface
+from redis.cache import CacheConfig, CacheInterface
 from redis.commands import (
     CoreCommands,
     RedisModuleCommands,
@@ -142,12 +142,10 @@ class initializer. In the case of conflicting arguments, querystring
 
         """
         single_connection_client = kwargs.pop("single_connection_client", False)
-        use_cache = kwargs.pop("use_cache", False)
         connection_pool = ConnectionPool.from_url(url, **kwargs)
         client = cls(
             connection_pool=connection_pool,
             single_connection_client=single_connection_client,
-            use_cache=use_cache,
         )
         client.auto_close_connection_pool = True
         return client
@@ -213,9 +211,8 @@ def __init__(
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
         protocol: Optional[int] = 2,
-        use_cache: bool = False,
         cache: Optional[CacheInterface] = None,
-        cache_config: Optional[CacheConfiguration] = None,
+        cache_config: Optional[CacheConfig] = None,
     ) -> None:
         """
         Initialize a new Redis client.
@@ -308,10 +305,9 @@ def __init__(
                             "ssl_ciphers": ssl_ciphers,
                         }
                     )
-                if use_cache and protocol in [3, "3"]:
+                if (cache_config or cache) and protocol in [3, "3"]:
                     kwargs.update(
                         {
-                            "use_cache": use_cache,
                             "cache": cache,
                             "cache_config": cache_config,
                         }
@@ -323,7 +319,7 @@ def __init__(
 
         self.connection_pool = connection_pool
 
-        if use_cache and self.connection_pool.get_protocol() not in [3, "3"]:
+        if (cache_config or cache) and self.connection_pool.get_protocol() not in [3, "3"]:
             raise RedisError("Client caching is only supported with RESP version 3")
 
         self.connection = None

redis/cluster.py

@@ -9,7 +9,7 @@
 from redis._parsers import CommandsParser, Encoder
 from redis._parsers.helpers import parse_scan
 from redis.backoff import default_backoff
-from redis.cache import CacheConfiguration, CacheInterface
+from redis.cache import CacheConfig, CacheInterface
 from redis.client import CaseInsensitiveDict, PubSub, Redis
 from redis.commands import READ_COMMANDS, RedisClusterCommands
 from redis.commands.helpers import list_or_args
@@ -168,7 +168,6 @@ def parse_cluster_myshardid(resp, **options):
     "ssl_password",
     "unix_socket_path",
     "username",
-    "use_cache",
     "cache",
     "cache_config",
 )
@@ -504,9 +503,8 @@ def __init__(
         dynamic_startup_nodes: bool = True,
         url: Optional[str] = None,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
-        use_cache: bool = False,
         cache: Optional[CacheInterface] = None,
-        cache_config: Optional[CacheConfiguration] = None,
+        cache_config: Optional[CacheConfig] = None,
         **kwargs,
     ):
         """
@@ -631,7 +629,7 @@ def __init__(
             kwargs.get("decode_responses", False),
         )
         protocol = kwargs.get("protocol", None)
-        if use_cache and protocol not in [3, "3"]:
+        if (cache_config or cache) and protocol not in [3, "3"]:
             raise RedisError("Client caching is only supported with RESP version 3")
 
         self.cluster_error_retry_attempts = cluster_error_retry_attempts
@@ -646,7 +644,6 @@ def __init__(
             require_full_coverage=require_full_coverage,
             dynamic_startup_nodes=dynamic_startup_nodes,
             address_remap=address_remap,
-            use_cache=use_cache,
             cache=cache,
             cache_config=cache_config,
             **kwargs,
@@ -1328,9 +1325,8 @@ def __init__(
         dynamic_startup_nodes=True,
         connection_pool_class=ConnectionPool,
         address_remap: Optional[Callable[[Tuple[str, int]], Tuple[str, int]]] = None,
-        use_cache: bool = False,
         cache: Optional[CacheInterface] = None,
-        cache_config: Optional[CacheConfiguration] = None,
+        cache_config: Optional[CacheConfig] = None,
         **kwargs,
     ):
         self.nodes_cache = {}
@@ -1343,7 +1339,6 @@ def __init__(
         self._dynamic_startup_nodes = dynamic_startup_nodes
         self.connection_pool_class = connection_pool_class
         self.address_remap = address_remap
-        self.use_cache = use_cache
         self.cache = cache
         self.cache_config = cache_config
         self._moved_exception = None
@@ -1489,15 +1484,13 @@ def create_redis_node(self, host, port, **kwargs):
             # Create a redis node with a costumed connection pool
             kwargs.update({"host": host})
             kwargs.update({"port": port})
-            kwargs.update({"use_cache": self.use_cache})
             kwargs.update({"cache": self.cache})
             kwargs.update({"cache_config": self.cache_config})
             r = Redis(connection_pool=self.connection_pool_class(**kwargs))
         else:
             r = Redis(
                 host=host,
                 port=port,
-                use_cache=self.use_cache,
                 cache=self.cache,
                 cache_config=self.cache_config,
                 **kwargs,

redis/connection.py

@@ -13,7 +13,7 @@
 from urllib.parse import parse_qs, unquote, urlparse
 
 from redis.cache import (
-    CacheConfiguration,
+    CacheConfig,
     CacheEntry,
     CacheEntryStatus,
     CacheFactory,
@@ -879,7 +879,7 @@ def _enable_tracking_callback(self, conn: ConnectionInterface) -> None:
         conn._parser.set_invalidation_push_handler(self._on_invalidation_callback)
 
     def _process_pending_invalidations(self):
-        while self.retry.call_with_retry_on_false(lambda: self.can_read()):
+        while self.can_read():
             self._conn.read_response(push_request=True)
 
     def _on_invalidation_callback(self, data: List[Union[str, Optional[List[bytes]]]]):
@@ -1251,13 +1251,12 @@ def __init__(
         self.connection_kwargs = connection_kwargs
         self.max_connections = max_connections
         self.cache = None
-        self._cache_conf = None
         self._cache_factory = cache_factory
         self._scheduler = None
         self._hc_cancel_event = None
         self._hc_thread = None
 
-        if connection_kwargs.get("use_cache"):
+        if connection_kwargs.get("cache_config") or connection_kwargs.get("cache"):
             if connection_kwargs.get("protocol") not in [3, "3"]:
                 raise RedisError("Client caching is only supported with RESP version 3")
 
@@ -1278,7 +1277,6 @@ def __init__(
 
             self._scheduler = Scheduler()
 
-        connection_kwargs.pop("use_cache", None)
         connection_kwargs.pop("cache", None)
         connection_kwargs.pop("cache_config", None)
 
@@ -1494,8 +1492,9 @@ def run_scheduled_healthcheck(self) -> None:
         # Run scheduled healthcheck to avoid stale invalidations in idle connections.
         if self.cache is not None and self._scheduler is not None:
             self._hc_cancel_event = threading.Event()
+            hc_interval = self.cache.get_config().get_health_check_interval()
             self._hc_thread = self._scheduler.run_with_interval(
-                self._perform_health_check, 2, self._hc_cancel_event
+                self._perform_health_check, hc_interval, self._hc_cancel_event
             )
 
     def _perform_health_check(self, done: threading.Event) -> None:

redis/retry.py

@@ -78,35 +78,3 @@ def call_with_retry(
                 backoff = self._backoff.compute(failures)
                 if backoff > 0:
                     sleep(backoff)
-
-    def call_with_retry_on_false(
-        self,
-        do: Callable[[], T],
-        on_false: Optional[Callable[[], T]] = None,
-        max_retries: Optional[int] = 3,
-        timeout: Optional[float] = 0,
-        exponent: Optional[int] = 2,
-    ) -> bool:
-        """
-        Execute an operation that returns boolean value with retry
-        logic in case if false value been returned.
-        `do`: the operation to call. Expects no argument.
-        `on_false`: Callback to be executed on retry fail.
-        """
-        res = do()
-
-        if res:
-            return res
-
-        if on_false is not None:
-            on_false()
-
-        if max_retries > 0:
-            if timeout > 0:
-                time.sleep(timeout)
-
-            return self.call_with_retry_on_false(
-                do, on_false, max_retries - 1, timeout * exponent, exponent
-            )
-
-        return False