Merge branch 'master' into 4.2.0rc2

Author: chayim

CHANGES

@@ -1,6 +1,9 @@
+
+    * Add `items` parameter to `hset` signature
   	* Create codeql-analysis.yml (#1988). Thanks @chayim
     * Add limited support for Lua scripting with RedisCluster
     * Implement `.lock()` method on RedisCluster
+
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)
     * Add redis5 and redis4 dockers (#1871)

redis/cluster.py

@@ -228,6 +228,7 @@ class RedisCluster(RedisClusterCommands):
                 "ACL SETUSER",
                 "ACL USERS",
                 "ACL WHOAMI",
+                "AUTH",
                 "CLIENT LIST",
                 "CLIENT SETNAME",
                 "CLIENT GETNAME",
@@ -283,19 +284,29 @@ class RedisCluster(RedisClusterCommands):
                 "READONLY",
                 "READWRITE",
                 "TIME",
+                "GRAPH.CONFIG",
             ],
             DEFAULT_NODE,
         ),
         list_keys_to_dict(
             [
                 "FLUSHALL",
                 "FLUSHDB",
+                "FUNCTION DELETE",
+                "FUNCTION FLUSH",
+                "FUNCTION LIST",
+                "FUNCTION LOAD",
+                "FUNCTION RESTORE",
                 "SCRIPT EXISTS",
                 "SCRIPT FLUSH",
                 "SCRIPT LOAD",
             ],
             PRIMARIES,
         ),
+        list_keys_to_dict(
+            ["FUNCTION DUMP"],
+            RANDOM,
+        ),
         list_keys_to_dict(
             [
                 "CLUSTER COUNTKEYSINSLOT",
@@ -809,6 +820,10 @@ def lock(
             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)
@@ -910,6 +925,10 @@ def determine_slot(self, *args):
         else:
             keys = self._get_command_keys(*args)
             if keys is None or len(keys) == 0:
+                # FCALL can call a function with 0 keys, that means the function
+                #  can be run on any node so we can just return a random slot
+                if command in ("FCALL", "FCALL_RO"):
+                    return random.randrange(0, REDIS_CLUSTER_HASH_SLOTS)
                 raise RedisClusterException(
                     "No way to dispatch this command to Redis Cluster. "
                     "Missing key.\nYou can execute the command by specifying "
@@ -1180,6 +1199,20 @@ def _process_result(self, command, res, **kwargs):
         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):
@@ -2025,7 +2058,13 @@ def _send_cluster_commands(
 
         # 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)
@@ -2039,6 +2078,9 @@ def _fail_on_redirect(self, allow_redirections):
                 "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")

redis/commands/cluster.py

@@ -4,11 +4,13 @@
 from .core import (
     ACLCommands,
     DataAccessCommands,
+    FunctionCommands,
     ManagementCommands,
     PubSubCommands,
     ScriptCommands,
 )
 from .helpers import list_or_args
+from .redismodules import RedisModuleCommands
 
 
 class ClusterMultiKeyCommands:
@@ -212,6 +214,8 @@ class RedisClusterCommands(
     PubSubCommands,
     ClusterDataAccessCommands,
     ScriptCommands,
+    FunctionCommands,
+    RedisModuleCommands,
 ):
     """
     A class for all Redis Cluster commands

redis/commands/core.py

@@ -369,14 +369,16 @@ class ManagementCommands(CommandsProtocol):
     Redis management commands
     """
 
-    def auth(self):
+    def auth(self, password, username=None, **kwargs):
         """
-        This function throws a NotImplementedError since it is intentionally
-        not supported.
+        Authenticates the user. If you do not pass username, Redis will try to
+        authenticate for the "default" user. If you do pass username, it will
+        authenticate for the given user.
+        For more information check https://redis.io/commands/auth
         """
-        raise NotImplementedError(
-            "AUTH is intentionally not implemented in the client."
-        )
+        if username:
+            return self.execute_command("AUTH", username, password, **kwargs)
+        return self.execute_command
 
     def bgrewriteaof(self, **kwargs):
         """Tell the Redis server to rewrite the AOF file from data in memory.
@@ -741,6 +743,15 @@ def command_info(self, **kwargs) -> None:
     def command_count(self, **kwargs) -> ResponseT:
         return self.execute_command("COMMAND COUNT", **kwargs)
 
+    def command_docs(self, *args):
+        """
+        This function throws a NotImplementedError since it is intentionally
+        not supported.
+        """
+        raise NotImplementedError(
+            "COMMAND DOCS is intentionally not implemented in the client."
+        )
+
     def config_get(self, pattern: PatternT = "*", **kwargs) -> ResponseT:
         """
         Return a dictionary of configuration based on the ``pattern``
@@ -1512,6 +1523,15 @@ def expireat(self, name: KeyT, when: AbsExpiryT) -> ResponseT:
             when = int(time.mktime(when.timetuple()))
         return self.execute_command("EXPIREAT", name, when)
 
+    def expiretime(self, key: str) -> int:
+        """
+        Returns the absolute Unix timestamp (since January 1, 1970) in seconds
+        at which the given key will expire.
+
+        For more information check https://redis.io/commands/expiretime
+        """
+        return self.execute_command("EXPIRETIME", key)
+
     def get(self, name: KeyT) -> ResponseT:
         """
         Return the value at key ``name``, or None if the key doesn't exist
@@ -1780,6 +1800,15 @@ def pexpireat(self, name: KeyT, when: AbsExpiryT) -> ResponseT:
             when = int(time.mktime(when.timetuple())) * 1000 + ms
         return self.execute_command("PEXPIREAT", name, when)
 
+    def pexpiretime(self, key: str) -> int:
+        """
+        Returns the absolute Unix timestamp (since January 1, 1970) in milliseconds
+        at which the given key will expire.
+
+        For more information check https://redis.io/commands/pexpiretime
+        """
+        return self.execute_command("PEXPIRETIME", key)
+
     def psetex(
         self,
         name: KeyT,
@@ -4587,18 +4616,21 @@ def hset(
         key: Optional[str] = None,
         value: Optional[str] = None,
         mapping: Optional[dict] = None,
+        items: Optional[list] = None,
     ) -> Union[Awaitable[int], int]:
         """
         Set ``key`` to ``value`` within hash ``name``,
         ``mapping`` accepts a dict of key/value pairs that will be
         added to hash ``name``.
+        ``items`` accepts a list of key/value pairs that will be
+        added to hash ``name``.
         Returns the number of fields that were added.
 
         For more information check https://redis.io/commands/hset
         """
-        if key is None and not mapping:
+        if key is None and not mapping and not items:
             raise DataError("'hset' with no key value pairs")
-        items = []
+        items = items or []
         if key is not None:
             items.extend((key, value))
         if mapping:

redis/commands/json/__init__.py

@@ -103,16 +103,34 @@ def pipeline(self, transaction=True, shard_hint=None):
         pipe.jsonget('foo')
         pipe.jsonget('notakey')
         """
-        p = Pipeline(
-            connection_pool=self.client.connection_pool,
-            response_callbacks=self.MODULE_CALLBACKS,
-            transaction=transaction,
-            shard_hint=shard_hint,
-        )
+        if isinstance(self.client, redis.RedisCluster):
+            p = ClusterPipeline(
+                nodes_manager=self.client.nodes_manager,
+                commands_parser=self.client.commands_parser,
+                startup_nodes=self.client.nodes_manager.startup_nodes,
+                result_callbacks=self.client.result_callbacks,
+                cluster_response_callbacks=self.client.cluster_response_callbacks,
+                cluster_error_retry_attempts=self.client.cluster_error_retry_attempts,
+                read_from_replicas=self.client.read_from_replicas,
+                reinitialize_steps=self.client.reinitialize_steps,
+            )
+
+        else:
+            p = Pipeline(
+                connection_pool=self.client.connection_pool,
+                response_callbacks=self.MODULE_CALLBACKS,
+                transaction=transaction,
+                shard_hint=shard_hint,
+            )
+
         p._encode = self._encode
         p._decode = self._decode
         return p
 
 
+class ClusterPipeline(JSONCommands, redis.cluster.ClusterPipeline):
+    """Cluster pipeline for the module."""
+
+
 class Pipeline(JSONCommands, redis.client.Pipeline):
     """Pipeline for the module."""

redis/commands/parser.py

@@ -17,7 +17,14 @@ def __init__(self, redis_connection):
         self.initialize(redis_connection)
 
     def initialize(self, r):
-        self.commands = r.execute_command("COMMAND")
+        commands = r.execute_command("COMMAND")
+        uppercase_commands = []
+        for cmd in commands:
+            if any(x.isupper() for x in cmd):
+                uppercase_commands.append(cmd)
+        for cmd in uppercase_commands:
+            commands[cmd.lower()] = commands.pop(cmd)
+        self.commands = commands
 
     # As soon as this PR is merged into Redis, we should reimplement
     # our logic to use COMMAND INFO changes to determine the key positions

redis/commands/search/__init__.py

@@ -1,3 +1,5 @@
+import redis
+
 from .commands import SearchCommands
 
 
@@ -17,7 +19,7 @@ def __init__(self, client, chunk_size=1000):
 
             self.client = client
             self.execute_command = client.execute_command
-            self.pipeline = client.pipeline(transaction=False, shard_hint=None)
+            self._pipeline = client.pipeline(transaction=False, shard_hint=None)
             self.total = 0
             self.chunk_size = chunk_size
             self.current_chunk = 0
@@ -42,7 +44,7 @@ def add_document(
             """
             self.client._add_document(
                 doc_id,
-                conn=self.pipeline,
+                conn=self._pipeline,
                 nosave=nosave,
                 score=score,
                 payload=payload,
@@ -67,7 +69,7 @@ def add_document_hash(
             """
             self.client._add_document_hash(
                 doc_id,
-                conn=self.pipeline,
+                conn=self._pipeline,
                 score=score,
                 replace=replace,
             )
@@ -80,7 +82,7 @@ def commit(self):
             """
             Manually commit and flush the batch indexing query
             """
-            self.pipeline.execute()
+            self._pipeline.execute()
             self.current_chunk = 0
 
     def __init__(self, client, index_name="idx"):
@@ -90,7 +92,25 @@ def __init__(self, client, index_name="idx"):
 
         If conn is not None, we employ an already existing redis connection
         """
+        self.MODULE_CALLBACKS = {}
         self.client = client
         self.index_name = index_name
         self.execute_command = client.execute_command
-        self.pipeline = client.pipeline
+        self._pipeline = client.pipeline
+
+    def pipeline(self, transaction=True, shard_hint=None):
+        """Creates a pipeline for the SEARCH module, that can be used for executing
+        SEARCH commands, as well as classic core commands.
+        """
+        p = Pipeline(
+            connection_pool=self.client.connection_pool,
+            response_callbacks=self.MODULE_CALLBACKS,
+            transaction=transaction,
+            shard_hint=shard_hint,
+        )
+        p.index_name = self.index_name
+        return p
+
+
+class Pipeline(SearchCommands, redis.client.Pipeline):
+    """Pipeline for the module."""