Merge remote-tracking branch 'original/master'

Author: utkarshgupta137

CHANGES

@@ -1,4 +1,5 @@
 
+    * Allow negative `retries` for `Retry` class to retry forever
     * Add `items` parameter to `hset` signature
     * Create codeql-analysis.yml (#1988). Thanks @chayim
     * Add limited support for Lua scripting with RedisCluster
@@ -7,6 +8,7 @@
     * Fix scan_iter for RedisCluster
     * Remove verbose logging when initializing ClusterPubSub, ClusterPipeline or RedisCluster
     * Fix broken connection writer lock-up for asyncio (#2065)
+    * Fix auth bug when provided with no username (#2086)
 
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)

redis/asyncio/client.py

@@ -41,8 +41,8 @@
 )
 from redis.commands import (
     AsyncCoreCommands,
+    AsyncRedisModuleCommands,
     AsyncSentinelCommands,
-    RedisModuleCommands,
     list_or_args,
 )
 from redis.compat import Protocol, TypedDict
@@ -81,7 +81,7 @@ async def __call__(self, response: Any, **kwargs):
 
 
 class Redis(
-    AbstractRedis, RedisModuleCommands, AsyncCoreCommands, AsyncSentinelCommands
+    AbstractRedis, AsyncRedisModuleCommands, AsyncCoreCommands, AsyncSentinelCommands
 ):
     """
     Implementation of the Redis protocol.

redis/asyncio/retry.py

@@ -27,6 +27,7 @@ def __init__(
         """
         Initialize a `Retry` object with a `Backoff` object
         that retries a maximum of `retries` times.
+        `retries` can be negative to retry forever.
         You can specify the types of supported errors which trigger
         a retry with the `supported_errors` parameter.
         """
@@ -51,7 +52,7 @@ async def call_with_retry(
             except self._supported_errors as error:
                 failures += 1
                 await fail(error)
-                if failures > self._retries:
+                if self._retries >= 0 and failures > self._retries:
                     raise error
                 backoff = self._backoff.compute(failures)
                 if backoff > 0:

redis/cluster.py

@@ -802,7 +802,10 @@ def set_response_callback(self, command, callback):
         self.cluster_response_callbacks[command] = callback
 
     def _determine_nodes(self, *args, **kwargs):
-        command = args[0]
+        command = args[0].upper()
+        if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
+            command = f"{args[0]} {args[1]}".upper()
+
         nodes_flag = kwargs.pop("nodes_flag", None)
         if nodes_flag is not None:
             # nodes flag passed by the user

redis/commands/__init__.py

@@ -2,7 +2,7 @@
 from .core import AsyncCoreCommands, CoreCommands
 from .helpers import list_or_args
 from .parser import CommandsParser
-from .redismodules import RedisModuleCommands
+from .redismodules import AsyncRedisModuleCommands, RedisModuleCommands
 from .sentinel import AsyncSentinelCommands, SentinelCommands
 
 __all__ = [
@@ -12,6 +12,7 @@
     "AsyncCoreCommands",
     "CoreCommands",
     "list_or_args",
+    "AsyncRedisModuleCommands",
     "RedisModuleCommands",
     "AsyncSentinelCommands",
     "SentinelCommands",

redis/commands/cluster.py

@@ -315,19 +315,19 @@ def replicaof(self, *args, **kwargs) -> NoReturn:
 
         For more information see https://redis.io/commands/replicaof
         """
-        raise RedisClusterException("REPLICAOF is not supported in cluster" " mode")
+        raise RedisClusterException("REPLICAOF is not supported in cluster mode")
 
     def swapdb(self, *args, **kwargs) -> NoReturn:
         """
         Swaps two Redis databases.
 
         For more information see https://redis.io/commands/swapdb
         """
-        raise RedisClusterException("SWAPDB is not supported in cluster" " mode")
+        raise RedisClusterException("SWAPDB is not supported in cluster mode")
 
     def cluster_myid(self, target_node: "TargetNodesT") -> ResponseT:
         """
-        Returns the node’s id.
+        Returns the node's id.
 
         :target_node: 'ClusterNode'
             The node to execute the command on
@@ -805,7 +805,6 @@ class RedisClusterCommands(
     target specific nodes. By default, if target_nodes is not specified, the
     command will be executed on the default cluster node.
 
-
     :param :target_nodes: type can be one of the followings:
         - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
         - 'ClusterNode'
@@ -834,7 +833,6 @@ class AsyncRedisClusterCommands(
     target specific nodes. By default, if target_nodes is not specified, the
     command will be executed on the default cluster node.
 
-
     :param :target_nodes: type can be one of the followings:
         - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
         - 'ClusterNode'

redis/commands/core.py

@@ -376,9 +376,11 @@ def auth(self, password, username=None, **kwargs):
         authenticate for the given user.
         For more information see https://redis.io/commands/auth
         """
-        if username:
-            return self.execute_command("AUTH", username, password, **kwargs)
-        return self.execute_command
+        pieces = []
+        if username is not None:
+            pieces.append(username)
+        pieces.append(password)
+        return self.execute_command("AUTH", *pieces, **kwargs)
 
     def bgrewriteaof(self, **kwargs):
         """Tell the Redis server to rewrite the AOF file from data in memory.
@@ -3817,7 +3819,7 @@ def xrevrange(
     def xtrim(
         self,
         name: KeyT,
-        maxlen: int,
+        maxlen: Union[int, None],
         approximate: bool = True,
         minid: Union[StreamIdT, None] = None,
         limit: Union[int, None] = None,

redis/commands/redismodules.py

@@ -81,3 +81,13 @@ def graph(self, index_name="idx"):
 
         g = Graph(client=self, name=index_name)
         return g
+
+
+class AsyncRedisModuleCommands(RedisModuleCommands):
+    def ft(self, index_name="idx"):
+        """Access the search namespace, providing support for redis search."""
+
+        from .search import AsyncSearch
+
+        s = AsyncSearch(client=self, index_name=index_name)
+        return s

redis/commands/search/__init__.py

@@ -1,6 +1,6 @@
 import redis
 
-from .commands import SearchCommands
+from .commands import AsyncSearchCommands, SearchCommands
 
 
 class Search(SearchCommands):
@@ -112,5 +112,67 @@ def pipeline(self, transaction=True, shard_hint=None):
         return p
 
 
+class AsyncSearch(Search, AsyncSearchCommands):
+    class BatchIndexer(Search.BatchIndexer):
+        """
+        A batch indexer allows you to automatically batch
+        document indexing in pipelines, flushing it every N documents.
+        """
+
+        async def add_document(
+            self,
+            doc_id,
+            nosave=False,
+            score=1.0,
+            payload=None,
+            replace=False,
+            partial=False,
+            no_create=False,
+            **fields,
+        ):
+            """
+            Add a document to the batch query
+            """
+            self.client._add_document(
+                doc_id,
+                conn=self._pipeline,
+                nosave=nosave,
+                score=score,
+                payload=payload,
+                replace=replace,
+                partial=partial,
+                no_create=no_create,
+                **fields,
+            )
+            self.current_chunk += 1
+            self.total += 1
+            if self.current_chunk >= self.chunk_size:
+                await self.commit()
+
+        async def commit(self):
+            """
+            Manually commit and flush the batch indexing query
+            """
+            await self._pipeline.execute()
+            self.current_chunk = 0
+
+    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 = AsyncPipeline(
+            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."""
+
+
+class AsyncPipeline(AsyncSearchCommands, redis.asyncio.client.Pipeline):
+    """AsyncPipeline for the module."""