Add async client-side caching support

docs/resp3_features.rst

@@ -86,10 +86,11 @@ Client-side caching
 Client-side caching is a technique used to create high performance services.
 It utilizes the memory on application servers, typically separate from the database nodes, to cache a subset of the data directly on the application side.
 For more information please check the `Redis client-side caching documentation <https://redis.io/docs/latest/develop/use/client-side-caching/>`_.
-Please notice that this feature is available only with RESP3 protocol enabled
-in sync clients. redis-py 8.0 and later use RESP3 on the wire by default, and
-the examples below pass ``protocol=3`` explicitly to make the requirement clear.
-Supported in standalone, Cluster, and Sentinel clients.
+Please notice that this feature is available only with RESP3 protocol enabled.
+redis-py 8.0 and later use RESP3 on the wire by default, and the examples below
+pass ``protocol=3`` explicitly to make the requirement clear. It is supported
+by standalone sync and async clients, as well as the sync Cluster and Sentinel
+clients.
 
 Basic usage:
 
@@ -101,7 +102,15 @@ Enable caching with default configuration:
     >>> from redis.cache import CacheConfig
     >>> r = redis.Redis(host='localhost', port=6379, protocol=3, cache_config=CacheConfig())
 
-The same interface applies to Redis Cluster and Sentinel.
+The async client uses the same configuration:
+
+.. code:: python
+
+    >>> import redis.asyncio as redis
+    >>> from redis.cache import CacheConfig
+    >>> r = redis.Redis(host='localhost', port=6379, protocol=3, cache_config=CacheConfig())
+
+The same interface applies to the sync Redis Cluster and Sentinel clients.
 
 Enable caching with custom cache implementation:
 

redis/asyncio/client.py

@@ -52,6 +52,7 @@
 )
 from redis.asyncio.retry import Retry
 from redis.backoff import ExponentialWithJitterBackoff
+from redis.cache import CacheConfig, CacheInterface
 from redis.client import (
     EMPTY_RESPONSE,
     NEVER_DECODE,
@@ -316,6 +317,8 @@ def __init__(
         credential_provider: CredentialProvider | None = None,
         protocol: int | None = None,
         legacy_responses: bool = True,
+        cache: CacheInterface | None = None,
+        cache_config: CacheConfig | None = None,
         event_dispatcher: EventDispatcher | None = None,
         maint_notifications_config: MaintNotificationsConfig | None = None,
     ):
@@ -461,6 +464,13 @@ def __init__(
                             "ssl_password": ssl_password,
                         }
                     )
+                if (cache_config or cache) and check_protocol_version(protocol, 3):
+                    kwargs.update(
+                        {
+                            "cache": cache,
+                            "cache_config": cache_config,
+                        }
+                    )
             maint_notifications_enabled = (
                 maint_notifications_config and maint_notifications_config.enabled
             )
@@ -492,6 +502,12 @@ def __init__(
             )
 
         self.connection_pool = connection_pool
+
+        if (cache_config or cache) and not check_protocol_version(
+            self.connection_pool.get_protocol(), 3
+        ):
+            raise RedisError("Client caching is only supported with RESP version 3")
+
         self.single_connection_client = single_connection_client
         self.connection: Optional[Connection] = None
 
@@ -544,6 +560,10 @@ def get_encoder(self):
         """Get the connection pool's encoder"""
         return self.connection_pool.get_encoder()
 
+    def get_cache(self) -> Optional[CacheInterface]:
+        """Return the client-side cache configured for this connection pool."""
+        return self.connection_pool.cache
+
     def get_connection_kwargs(self):
         """Get the connection's key-word arguments"""
         return self.connection_pool.connection_kwargs
@@ -867,7 +887,7 @@ async def _send_command_parse_response(self, conn, command_name, *args, **option
             # returns its arity error instead of a client-side IndexError here.
             key, fieldset_name, values = himport_set
             return await self._himport_execute_set(conn, key, fieldset_name, values)
-        await conn.send_command(*args)
+        await conn.send_command(*args, **options)
         return await self.parse_response(conn, command_name, **options)
 
     async def _himport_reconcile_discards(self, conn):