Merge branch 'master' into cluster-shards

Author: dvora-h

.github/workflows/integration.yaml

@@ -22,7 +22,7 @@ jobs:
      steps:
        - uses: actions/checkout@v2
        - name: install python
-         uses: actions/setup-python@v2
+         uses: actions/setup-python@v3
          with:
            python-version: 3.9
        - name: run code linters
@@ -32,20 +32,27 @@ jobs:
 
    run-tests:
      runs-on: ubuntu-latest
+     continue-on-error: ${{ matrix.experimental }}
      timeout-minutes: 30
      strategy:
        max-parallel: 15
        matrix:
          python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
          test-type: ['standalone', 'cluster']
          connection-type: ['hiredis', 'plain']
+         experimental: [false]
+         include:
+           - python-version: 3.11.0-alpha.6
+             experimental: true
+             test-type: standalone
+             connection-type: plain
      env:
        ACTIONS_ALLOW_UNSECURE_COMMANDS: true
      name: Python ${{ matrix.python-version }} ${{matrix.test-type}}-${{matrix.connection-type}} tests
      steps:
        - uses: actions/checkout@v2
        - name: install python
-         uses: actions/setup-python@v2
+         uses: actions/setup-python@v3
          with:
            python-version: ${{ matrix.python-version }}
        - name: run tests
@@ -68,23 +75,28 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - name: install python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: 3.9
       - name: Run installed unit tests
         run: |
           bash .github/workflows/install_and_test.sh ${{ matrix.extension }}
 
    install_package_from_commit:
+    continue-on-error: ${{ matrix.experimental }}
     name: Install package from commit hash
     runs-on: ubuntu-latest
     strategy:
       matrix:
         python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
+        experimental: [false]
+        include:
+          - python-version: 3.11.0-alpha.5
+          - experimental: true
     steps:
       - uses: actions/checkout@v2
       - name: install python ${{ matrix.python-version }}
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: ${{ matrix.python-version }}
       - name: install from pip

.github/workflows/pypi-publish.yaml

@@ -11,7 +11,7 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - name: install python
-        uses: actions/setup-python@v2
+        uses: actions/setup-python@v3
         with:
           python-version: 3.9
       - name: Install dev tools

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)

docker/redis7/master/redis.conf

@@ -0,0 +1,4 @@
+port 6379
+save ""
+enable-debug-command yes
+enable-module-command yes

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.
@@ -693,16 +693,24 @@ async def execute_command(self, *args: EncodableT):
         # legitimate message off the stack if the connection is already
         # subscribed to one or more channels
 
+        await self.connect()
+        connection = self.connection
+        kwargs = {"check_health": not self.subscribed}
+        await self._execute(connection, connection.send_command, *args, **kwargs)
+
+    async def connect(self):
+        """
+        Ensure that the PubSub is connected
+        """
         if self.connection is None:
             self.connection = await self.connection_pool.get_connection(
                 "pubsub", self.shard_hint
             )
             # register a callback that re-subscribes to any channels we
             # were listening to when we were disconnected
             self.connection.register_connect_callback(self.on_connect)
-        connection = self.connection
-        kwargs = {"check_health": not self.subscribed}
-        await self._execute(connection, connection.send_command, *args, **kwargs)
+        else:
+            await self.connection.connect()
 
     async def _disconnect_raise_connect(self, conn, error):
         """
@@ -962,6 +970,7 @@ async def run(
             if handler is None:
                 raise PubSubError(f"Pattern: '{pattern}' has no handler registered")
 
+        await self.connect()
         while True:
             try:
                 await self.get_message(

redis/asyncio/lock.py

@@ -3,7 +3,7 @@
 import threading
 import uuid
 from types import SimpleNamespace
-from typing import TYPE_CHECKING, Awaitable, NoReturn, Optional, Union
+from typing import TYPE_CHECKING, Awaitable, Optional, Union
 
 from redis.exceptions import LockError, LockNotOwnedError
 
@@ -243,15 +243,15 @@ async def owned(self) -> bool:
             stored_token = encoder.encode(stored_token)
         return self.local.token is not None and stored_token == self.local.token
 
-    def release(self) -> Awaitable[NoReturn]:
+    def release(self) -> Awaitable[None]:
         """Releases the already acquired lock"""
         expected_token = self.local.token
         if expected_token is None:
             raise LockError("Cannot release an unlocked lock")
         self.local.token = None
         return self.do_release(expected_token)
 
-    async def do_release(self, expected_token: bytes):
+    async def do_release(self, expected_token: bytes) -> None:
         if not bool(
             await self.lua_release(
                 keys=[self.name], args=[expected_token], client=self.redis

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/client.py

@@ -760,7 +760,6 @@ class AbstractRedis:
         "DEBUG OBJECT": parse_debug_object,
         "FUNCTION DELETE": bool_ok,
         "FUNCTION FLUSH": bool_ok,
-        "FUNCTION LOAD": bool_ok,
         "FUNCTION RESTORE": bool_ok,
         "GEOHASH": lambda r: list(map(str_if_bytes, r)),
         "GEOPOS": lambda r: list(

redis/cluster.py

@@ -886,7 +886,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__ = [
@@ -11,6 +11,7 @@
     "AsyncCoreCommands",
     "CoreCommands",
     "list_or_args",
+    "AsyncRedisModuleCommands",
     "RedisModuleCommands",
     "AsyncSentinelCommands",
     "SentinelCommands",

redis/commands/cluster.py

@@ -189,15 +189,15 @@ def replicaof(self, *args, **kwargs):
 
         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):
         """
         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")
 
 
 class ClusterDataAccessCommands(DataAccessCommands):
@@ -310,7 +310,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'
@@ -323,7 +322,7 @@ class RedisClusterCommands(
 
     def cluster_myid(self, target_node):
         """
-        Returns the node’s id.
+        Returns the node's id.
 
         :target_node: 'ClusterNode'
             The node to execute the command on