Client Side Caching: Alpha support

redis/__init__.py

@@ -2,6 +2,7 @@
 
 from redis import asyncio  # noqa
 from redis.backoff import default_backoff
+from redis.cache import _Cache
 from redis.client import Redis, StrictRedis
 from redis.cluster import RedisCluster
 from redis.connection import (
@@ -61,6 +62,7 @@ def int_or_str(value):
     VERSION = tuple([99, 99, 99])
 
 __all__ = [
+    "_Cache",
     "AuthenticationError",
     "AuthenticationWrongNumberOfArgsError",
     "BlockingConnectionPool",

redis/cache.py

@@ -0,0 +1,95 @@
+import random
+import time
+from collections import OrderedDict, defaultdict
+
+
+class _Cache:
+    def __init__(self, max_size: int, ttl: int, eviction_policy: str, **kwargs):
+        self.max_size = max_size
+        self.ttl = ttl
+        self.eviction_policy = eviction_policy
+        self.cache = OrderedDict()
+        self.key_commands_map = defaultdict(set)
+        self.commands_ttl_list = []
+
+    def set(self, command, response, keys_in_command):
+        if len(self.cache) >= self.max_size:
+            self._evict()
+        self.cache[command] = {
+            "response": response,
+            "keys": keys_in_command,
+            "created_time": time.monotonic(),
+        }
+        if self.eviction_policy == "lfu":
+            self.cache[command]["access_count"] = 0
+        self._update_key_commands_map(keys_in_command, command)
+        self.commands_ttl_list.append(command)
+
+    def get(self, command):
+        if command in self.cache:
+            if self._is_expired(command):
+                del self.cache[command]
+                keys_in_command = self.cache[command]["keys"]
+                self._del_key_commands_map(keys_in_command, command)
+                return None
+            self._update_access(command)
+            return self.cache[command]["response"]
+        return None
+
+    def delete(self, command):
+        if command in self.cache:
+            keys_in_command = self.cache[command]["keys"]
+            self._del_key_commands_map(keys_in_command, command)
+            self.commands_ttl_list.remove(command)
+            del self.cache[command]
+
+    def delete_many(self, commands):
+        pass
+
+    def flush(self):
+        self.cache.clear()
+        self.key_commands_map.clear()
+        self.commands_ttl_list = []
+
+    def _is_expired(self, command):
+        if self.ttl == 0:
+            return False
+        return time.monotonic() - self.cache[command]["created_time"] > self.ttl
+
+    def _update_access(self, command):
+        if self.eviction_policy == "lru":
+            self.cache.move_to_end(command)
+        elif self.eviction_policy == "lfu":
+            self.cache[command]["access_count"] = (
+                self.cache.get(command, {}).get("access_count", 0) + 1
+            )
+            self.cache.move_to_end(command)
+        elif self.eviction_policy == "random":
+            pass  # Random eviction doesn't require updates
+
+    def _evict(self):
+        if self._is_expired(self.commands_ttl_list[0]):
+            self.delete(self.commands_ttl_list[0])
+        elif self.eviction_policy == "lru":
+            self.cache.popitem(last=False)
+        elif self.eviction_policy == "lfu":
+            min_access_command = min(
+                self.cache, key=lambda k: self.cache[k].get("access_count", 0)
+            )
+            self.cache.pop(min_access_command)
+        elif self.eviction_policy == "random":
+            random_command = random.choice(list(self.cache.keys()))
+            self.cache.pop(random_command)
+
+    def _update_key_commands_map(self, keys, command):
+        for key in keys:
+            self.key_commands_map[key].add(command)
+
+    def _del_key_commands_map(self, keys, command):
+        for key in keys:
+            self.key_commands_map[key].remove(command)
+
+    def invalidate(self, key):
+        if key in self.key_commands_map:
+            for command in self.key_commands_map[key]:
+                self.delete(command)

redis/client.py

@@ -13,6 +13,7 @@
     _RedisCallbacksRESP3,
     bool_ok,
 )
+from redis.cache import _Cache
 from redis.commands import (
     CoreCommands,
     RedisModuleCommands,
@@ -33,6 +34,8 @@
 from redis.lock import Lock
 from redis.retry import Retry
 from redis.utils import (
+    DEFAULT_BLACKLIST,
+    DEFAULT_WHITELIST,
     HIREDIS_AVAILABLE,
     _set_info_logger,
     get_lib_version,
@@ -203,6 +206,13 @@ def __init__(
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
         protocol: Optional[int] = 2,
+        client_caching: bool = False,
+        client_cache: Optional[_Cache] = None,
+        cache_max_size: int = 100,
+        cache_ttl: int = 0,
+        cache_eviction_policy: str = "lru",
+        cache_blacklist: List[str] = DEFAULT_BLACKLIST,
+        cache_whitelist: List[str] = DEFAULT_WHITELIST,
     ) -> None:
         """
         Initialize a new Redis client.
@@ -310,6 +320,12 @@ def __init__(
         else:
             self.response_callbacks.update(_RedisCallbacksRESP2)
 
+        self.client_cache = client_cache
+        self.cache_blacklist = cache_blacklist
+        self.cache_whitelist = cache_whitelist
+        if client_caching:
+            self.client_cache = _Cache(cache_max_size, cache_ttl, cache_eviction_policy)
+
     def __repr__(self) -> str:
         return f"{type(self).__name__}<{repr(self.connection_pool)}>"
 
@@ -525,23 +541,50 @@ def _disconnect_raise(self, conn, error):
         ):
             raise error
 
+    def get_from_local_cache(self, command):
+        if (
+            self.client_cache is None
+            or command[0] in self.cache_blacklist
+            or command[0] not in self.cache_whitelist
+        ):
+            return None
+        return self.client_cache.get(command)
+
+    def add_to_local_cache(self, command, response, keys):
+        if (
+            self.client_cache is not None
+            and (self.cache_blacklist == [] or command[0] not in self.cache_blacklist)
+            and (self.cache_whitelist == [] or command[0] in self.cache_whitelist)
+        ):
+            self.client_cache.set(command, response, keys)
+
+    def delete_from_local_cache(self, command):
+        if self.client_cache is not None:
+            self.client_cache.delete(command)
+
     # COMMAND EXECUTION AND PROTOCOL PARSING
     def execute_command(self, *args, **options):
         """Execute a command and return a parsed response"""
-        pool = self.connection_pool
         command_name = args[0]
-        conn = self.connection or pool.get_connection(command_name, **options)
+        keys = options.pop("keys", None)
+        response_from_cache = self.get_from_local_cache(args)
+        if response_from_cache is not None:
+            return response_from_cache
+        else:
+            pool = self.connection_pool
+            conn = self.connection or pool.get_connection(command_name, **options)
 
-        try:
-            return conn.retry.call_with_retry(
-                lambda: self._send_command_parse_response(
-                    conn, command_name, *args, **options
-                ),
-                lambda error: self._disconnect_raise(conn, error),
-            )
-        finally:
-            if not self.connection:
-                pool.release(conn)
+            try:
+                response = conn.retry.call_with_retry(
+                    lambda: self._send_command_parse_response(
+                        conn, command_name, *args, **options
+                    ),
+                    lambda error: self._disconnect_raise(conn, error),
+                )
+                self.add_to_local_cache(args, response, keys)
+            finally:
+                if not self.connection:
+                    pool.release(conn)
 
     def parse_response(self, connection, command_name, **options):
         """Parses a response from the Redis server"""