Merge branch 'master' into cluster-fixes

Author: chayim

.github/workflows/integration.yaml

@@ -49,7 +49,7 @@ jobs:
      strategy:
        max-parallel: 15
        matrix:
-         python-version: ['3.7', '3.8', '3.9', '3.10', 'pypy-3.7', 'pypy-3.8']
+         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8']
          test-type: ['standalone', 'cluster']
          connection-type: ['hiredis', 'plain']
      env:
@@ -106,7 +106,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', 'pypy-3.7']
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7']
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4

README.md

@@ -49,22 +49,22 @@ True
 b'bar'
 ```
 
-The above code connects to localhost on port 6379, sets a value in Redis, and retrieves it. All responses are returned as bytes in Python, to receive decoded strings, set *decode_responses=True*.  For this, and more connection options, see [these examples](https://redis.readthedocs.io/en/stable/examples.html)
+The above code connects to localhost on port 6379, sets a value in Redis, and retrieves it. All responses are returned as bytes in Python, to receive decoded strings, set *decode_responses=True*.  For this, and more connection options, see [these examples](https://redis.readthedocs.io/en/stable/examples.html).
 
 ### Connection Pools
 
-By default, redis-py uses a connection pool to manage connections. Each instance of a Redis class receives its own connection pool. You can however define your own [redis.ConnectionPool](https://redis.readthedocs.io/en/stable/connections.html#connection-pools)
+By default, redis-py uses a connection pool to manage connections. Each instance of a Redis class receives its own connection pool. You can however define your own [redis.ConnectionPool](https://redis.readthedocs.io/en/stable/connections.html#connection-pools).
 
 ``` python
 >>> pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
 >>> r = redis.Redis(connection_pool=pool)
 ```
 
-Alternatively, you might want to look at [Async connections](https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html), or [Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#cluster-client), or even [Async Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#async-cluster-client)
+Alternatively, you might want to look at [Async connections](https://redis.readthedocs.io/en/stable/examples/asyncio_examples.html), or [Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#cluster-client), or even [Async Cluster connections](https://redis.readthedocs.io/en/stable/connections.html#async-cluster-client).
 
 ### Redis Commands
 
-There is built-in support for all of the [out-of-the-box Redis commands](https://redis.io/commands). They are exposed using the raw Redis command names (`HSET`, `HGETALL`, etc.) except where a word (i.e del) is reserved by the language. The complete set of commands can be found [here](https://github.com/redis/redis-py/tree/master/redis/commands), or [the documentation](https://redis.readthedocs.io/en/stable/commands.html).
+There is built-in support for all of the [out-of-the-box Redis commands](https://redis.io/commands). They are exposed using the raw Redis command names (`HSET`, `HGETALL`, etc.) except where a word (i.e. del) is reserved by the language. The complete set of commands can be found [here](https://github.com/redis/redis-py/tree/master/redis/commands), or [the documentation](https://redis.readthedocs.io/en/stable/commands.html).
 
 ## Advanced Topics
 

docs/examples/asyncio_examples.ipynb

@@ -140,38 +140,31 @@
    "source": [
     "import asyncio\n",
     "\n",
-    "import async_timeout\n",
-    "\n",
     "import redis.asyncio as redis\n",
     "\n",
     "STOPWORD = \"STOP\"\n",
     "\n",
     "\n",
     "async def reader(channel: redis.client.PubSub):\n",
     "    while True:\n",
-    "        try:\n",
-    "            async with async_timeout.timeout(1):\n",
-    "                message = await channel.get_message(ignore_subscribe_messages=True)\n",
-    "                if message is not None:\n",
-    "                    print(f\"(Reader) Message Received: {message}\")\n",
-    "                    if message[\"data\"].decode() == STOPWORD:\n",
-    "                        print(\"(Reader) STOP\")\n",
-    "                        break\n",
-    "                await asyncio.sleep(0.01)\n",
-    "        except asyncio.TimeoutError:\n",
-    "            pass\n",
+    "        message = await channel.get_message(ignore_subscribe_messages=True)\n",
+    "        if message is not None:\n",
+    "            print(f\"(Reader) Message Received: {message}\")\n",
+    "            if message[\"data\"].decode() == STOPWORD:\n",
+    "                print(\"(Reader) STOP\")\n",
+    "                break\n",
     "\n",
     "r = redis.from_url(\"redis://localhost\")\n",
-    "pubsub = r.pubsub()\n",
-    "await pubsub.subscribe(\"channel:1\", \"channel:2\")\n",
+    "async with r.pubsub() as pubsub:\n",
+    "    await pubsub.subscribe(\"channel:1\", \"channel:2\")\n",
     "\n",
-    "future = asyncio.create_task(reader(pubsub))\n",
+    "    future = asyncio.create_task(reader(pubsub))\n",
     "\n",
-    "await r.publish(\"channel:1\", \"Hello\")\n",
-    "await r.publish(\"channel:2\", \"World\")\n",
-    "await r.publish(\"channel:1\", STOPWORD)\n",
+    "    await r.publish(\"channel:1\", \"Hello\")\n",
+    "    await r.publish(\"channel:2\", \"World\")\n",
+    "    await r.publish(\"channel:1\", STOPWORD)\n",
     "\n",
-    "await future"
+    "    await future"
    ]
   },
   {
@@ -204,39 +197,32 @@
    "source": [
     "import asyncio\n",
     "\n",
-    "import async_timeout\n",
-    "\n",
     "import redis.asyncio as redis\n",
     "\n",
     "STOPWORD = \"STOP\"\n",
     "\n",
     "\n",
     "async def reader(channel: redis.client.PubSub):\n",
     "    while True:\n",
-    "        try:\n",
-    "            async with async_timeout.timeout(1):\n",
-    "                message = await channel.get_message(ignore_subscribe_messages=True)\n",
-    "                if message is not None:\n",
-    "                    print(f\"(Reader) Message Received: {message}\")\n",
-    "                    if message[\"data\"].decode() == STOPWORD:\n",
-    "                        print(\"(Reader) STOP\")\n",
-    "                        break\n",
-    "                await asyncio.sleep(0.01)\n",
-    "        except asyncio.TimeoutError:\n",
-    "            pass\n",
+    "        message = await channel.get_message(ignore_subscribe_messages=True)\n",
+    "        if message is not None:\n",
+    "            print(f\"(Reader) Message Received: {message}\")\n",
+    "            if message[\"data\"].decode() == STOPWORD:\n",
+    "                print(\"(Reader) STOP\")\n",
+    "                break\n",
     "\n",
     "\n",
     "r = await redis.from_url(\"redis://localhost\")\n",
-    "pubsub = r.pubsub()\n",
-    "await pubsub.psubscribe(\"channel:*\")\n",
+    "async with r.pubsub() as pubsub:\n",
+    "    await pubsub.psubscribe(\"channel:*\")\n",
     "\n",
-    "future = asyncio.create_task(reader(pubsub))\n",
+    "    future = asyncio.create_task(reader(pubsub))\n",
     "\n",
-    "await r.publish(\"channel:1\", \"Hello\")\n",
-    "await r.publish(\"channel:2\", \"World\")\n",
-    "await r.publish(\"channel:1\", STOPWORD)\n",
+    "    await r.publish(\"channel:1\", \"Hello\")\n",
+    "    await r.publish(\"channel:2\", \"World\")\n",
+    "    await r.publish(\"channel:1\", STOPWORD)\n",
     "\n",
-    "await future"
+    "    await future"
    ]
   },
   {
@@ -298,4 +284,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 1
-}
+}

redis/commands/cluster.py

@@ -316,6 +316,25 @@ async def _split_command_across_slots(self, command: str, *keys: KeyT) -> int:
         # Sum up the reply from each command
         return sum(await self._execute_pipeline_by_slot(command, slots_to_keys))
 
+    async def _execute_pipeline_by_slot(
+        self, command: str, slots_to_args: Mapping[int, Iterable[EncodableT]]
+    ) -> List[Any]:
+        if self._initialize:
+            await self.initialize()
+        read_from_replicas = self.read_from_replicas and command in READ_COMMANDS
+        pipe = self.pipeline()
+        [
+            pipe.execute_command(
+                command,
+                *slot_args,
+                target_nodes=[
+                    self.nodes_manager.get_node_from_slot(slot, read_from_replicas)
+                ],
+            )
+            for slot, slot_args in slots_to_args.items()
+        ]
+        return await pipe.execute()
+
 
 class ClusterManagementCommands(ManagementCommands):
     """

redis/commands/core.py

@@ -2553,7 +2553,7 @@ def lmpop(
         self,
         num_keys: int,
         *args: List[str],
-        direction: str = None,
+        direction: str,
         count: Optional[int] = 1,
     ) -> Union[Awaitable[list], list]:
         """

tests/test_asyncio/test_cluster.py

@@ -874,6 +874,15 @@ async def test_unlink(self, r: RedisCluster) -> None:
         await asyncio.sleep(0.1)
         assert await r.unlink(*d.keys()) == 0
 
+    async def test_initialize_before_execute_multi_key_command(
+        self, request: FixtureRequest
+    ) -> None:
+        # Test for issue https://github.com/redis/redis-py/issues/2437
+        url = request.config.getoption("--redis-url")
+        r = RedisCluster.from_url(url)
+        assert 0 == await r.exists("a", "b", "c")
+        await r.close()
+
     @skip_if_redis_enterprise()
     async def test_cluster_myid(self, r: RedisCluster) -> None:
         node = r.get_random_node()