Hacky test + fixes for it

Author: Alex Kouzemtchenko

redis/asyncio/cluster.py

@@ -1434,6 +1434,9 @@ def __init__(self, client: RedisCluster) -> None:
 
         self._command_stack: List["PipelineCommand"] = []
 
+        self.nodes_manager = self._client.nodes_manager
+        self.set_response_callback = self._client.set_response_callback
+
     async def initialize(self) -> "ClusterPipeline":
         if self._client._initialize:
             await self._client.initialize()

redis/commands/json/__init__.py

@@ -1,6 +1,7 @@
 from json import JSONDecodeError, JSONDecoder, JSONEncoder
 
 import redis
+import redis.asyncio as redis_asyncio
 
 from ..helpers import get_protocol_version, nativestr
 from .commands import JSONCommands
@@ -125,7 +126,10 @@ def pipeline(self, transaction=True, shard_hint=None):
                 reinitialize_steps=self.client.reinitialize_steps,
                 lock=self.client._lock,
             )
-
+        elif isinstance(self.client, redis_asyncio.RedisCluster):
+            p = AsyncClusterPipeline(
+                client = self.client
+            )
         else:
             p = Pipeline(
                 connection_pool=self.client.connection_pool,
@@ -143,5 +147,9 @@ class ClusterPipeline(JSONCommands, redis.cluster.ClusterPipeline):
     """Cluster pipeline for the module."""
 
 
+class AsyncClusterPipeline(JSONCommands, redis_asyncio.cluster.ClusterPipeline):
+    """Cluster pipeline for the module."""
+
+
 class Pipeline(JSONCommands, redis.client.Pipeline):
     """Pipeline for the module."""

redis/commands/redismodules.py

@@ -84,6 +84,14 @@ def graph(self, index_name="idx"):
 
 
 class AsyncRedisModuleCommands(RedisModuleCommands):
+    def json(self, encoder=JSONEncoder(), decoder=JSONDecoder()):
+        """Access the json namespace, providing support for redis json."""
+
+        from .json import JSON
+
+        jj = JSON(client=self, encoder=encoder, decoder=decoder)
+        return jj
+
     def ft(self, index_name="idx"):
         """Access the search namespace, providing support for redis search."""
 

setup.py

@@ -8,7 +8,7 @@
     long_description_content_type="text/markdown",
     keywords=["Redis", "key-value store", "database"],
     license="MIT",
-    version="5.1.0b3",
+    version="5.1.2",
     packages=find_packages(
         include=[
             "redis",

test.py

@@ -0,0 +1,20 @@
+import asyncio
+from redis.asyncio.cluster import RedisCluster  # type: ignore
+import redis.asyncio as redis  # type: ignore
+
+async def main():
+    r = RedisCluster.from_url("redis://localhost:16379/0")
+    p = r.pipeline()
+    p.json().set("blah", ".", 1)
+    p.set("blah2", 1)
+    await p.execute()
+
+
+    p = r.json().pipeline()
+    p.set("blah3", ".", 1)
+    await p.execute()
+    print("done")
+
+loop = asyncio.new_event_loop()
+asyncio.set_event_loop(loop)
+asyncio.get_event_loop().run_until_complete(main())