Add support for JSON, TIMESERIES, BLOOM & GRAPH commands in cluster (#2032)

Co-authored-by: Chayim <chayim@users.noreply.github.com>

1 files changed, 29 insertions, 1 deletions

diff --git a/redis/cluster.py b/redis/cluster.py
index 4491e29..4b327ad 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -284,6 +284,7 @@ class RedisCluster(RedisClusterCommands):
                 "READONLY",
                 "READWRITE",
                 "TIME",
+                "GRAPH.CONFIG",
             ],
             DEFAULT_NODE,
         ),
@@ -810,6 +811,10 @@ class RedisCluster(RedisClusterCommands):
             thread_local=thread_local,
         )
 
+    def set_response_callback(self, command, callback):
+        """Set a custom Response Callback"""
+        self.cluster_response_callbacks[command] = callback
+
     def _determine_nodes(self, *args, **kwargs):
         command = args[0]
         nodes_flag = kwargs.pop("nodes_flag", None)
@@ -1181,6 +1186,20 @@ class RedisCluster(RedisClusterCommands):
         else:
             return res
 
+    def load_external_module(
+        self,
+        funcname,
+        func,
+    ):
+        """
+        This function can be used to add externally defined redis modules,
+        and their namespaces to the redis client.
+
+        ``funcname`` - A string containing the name of the function to create
+        ``func`` - The function, being added to this class.
+        """
+        setattr(self, funcname, func)
+
 
 class ClusterNode:
     def __init__(self, host, port, server_type=None, redis_connection=None):
@@ -2026,7 +2045,13 @@ class ClusterPipeline(RedisCluster):
 
         # turn the response back into a simple flat array that corresponds
         # to the sequence of commands issued in the stack in pipeline.execute()
-        response = [c.result for c in sorted(stack, key=lambda x: x.position)]
+        response = []
+        for c in sorted(stack, key=lambda x: x.position):
+            if c.args[0] in self.cluster_response_callbacks:
+                c.result = self.cluster_response_callbacks[c.args[0]](
+                    c.result, **c.options
+                )
+            response.append(c.result)
 
         if raise_on_error:
             self.raise_first_error(stack)
@@ -2040,6 +2065,9 @@ class ClusterPipeline(RedisCluster):
                 "ASK & MOVED redirection not allowed in this pipeline"
             )
 
+    def exists(self, *keys):
+        return self.execute_command("EXISTS", *keys)
+
     def eval(self):
         """ """
         raise RedisClusterException("method eval() is not implemented")