Merge branch 'master' into ckpy312

Author: chayim

.github/workflows/spellcheck.yml

@@ -8,7 +8,7 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
       - name: Check Spelling
-        uses: rojopolis/spellcheck-github-actions@0.34.0
+        uses: rojopolis/spellcheck-github-actions@0.35.0
         with:
           config_path: .github/spellcheck-settings.yml
           task_name: Markdown

CHANGES

@@ -1,3 +1,6 @@
+    * Fix return types for `get`, `set_path` and `strappend` in JSONCommands
+    * Connection.register_connect_callback() is made public.
+    * Fix async `read_response` to use `disable_decoding`.
     * Add 'aclose()' methods to async classes, deprecate async close().
     * Fix #2831, add auto_close_connection_pool=True arg to asyncio.Redis.from_url()
     * Fix incorrect redis.asyncio.Cluster type hint for `retry_on_error`
@@ -55,6 +58,8 @@
     * Fix for Unhandled exception related to self.host with unix socket (#2496)
     * Improve error output for master discovery 
     * Make `ClusterCommandsProtocol` an actual Protocol
+    * Add `sum` to DUPLICATE_POLICY documentation of `TS.CREATE`, `TS.ADD` and `TS.ALTER`
+    * Prevent async ClusterPipeline instances from becoming "false-y" in case of empty command stack (#3061)
 
 * 4.1.3 (Feb 8, 2022)
   * Fix flushdb and flushall (#1926)

docs/advanced_features.rst

@@ -346,7 +346,7 @@ running.
 
 The third option runs an event loop in a separate thread.
 pubsub.run_in_thread() creates a new thread and starts the event loop.
-The thread object is returned to the caller of [un_in_thread(). The
+The thread object is returned to the caller of run_in_thread(). The
 caller can use the thread.stop() method to shut down the event loop and
 thread. Behind the scenes, this is simply a wrapper around get_message()
 that runs in a separate thread, essentially creating a tiny non-blocking

docs/clustering.rst

@@ -92,7 +92,7 @@ The ‘target_nodes’ parameter is explained in the following section,
    >>> # target-node: default-node
    >>> rc.ping()
 
-Specfiying Target Nodes
+Specifying Target Nodes
 -----------------------
 
 As mentioned above, all non key-based RedisCluster commands accept the

docs/conf.py

@@ -86,7 +86,7 @@
 
 # List of patterns, relative to source directory, that match files and
 # directories to ignore when looking for source files.
-exclude_patterns = ["_build", "**.ipynb_checkponts"]
+exclude_patterns = ["_build", "**.ipynb_checkpoints"]
 
 # The reST default role (used for this markup: `text`) to use for all
 # documents.

docs/examples/pipeline_examples.ipynb

@@ -123,7 +123,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "The responses of the three commands are stored in a list. In the above example, the two first boolean indicates that the `set` commands were successfull and the last element of the list is the result of the `get(\"a\")` comand."
+    "The responses of the three commands are stored in a list. In the above example, the two first boolean indicates that the `set` commands were successful and the last element of the list is the result of the `get(\"a\")` comand."
    ]
   },
   {

redis/_parsers/helpers.py

@@ -322,7 +322,7 @@ def float_or_none(response):
     return float(response)
 
 
-def bool_ok(response):
+def bool_ok(response, **options):
     return str_if_bytes(response) == "OK"
 
 

redis/_parsers/hiredis.py

@@ -198,10 +198,16 @@ async def read_response(
         if not self._connected:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
 
-        response = self._reader.gets()
+        if disable_decoding:
+            response = self._reader.gets(False)
+        else:
+            response = self._reader.gets()
         while response is False:
             await self.read_from_socket()
-            response = self._reader.gets()
+            if disable_decoding:
+                response = self._reader.gets(False)
+            else:
+                response = self._reader.gets()
 
         # if the response is a ConnectionError or the response is a list and
         # the first item is a ConnectionError, raise it as something bad

redis/_parsers/resp3.py

@@ -6,15 +6,18 @@
 from .base import _AsyncRESPBase, _RESPBase
 from .socket import SERVER_CLOSED_CONNECTION_ERROR
 
+_INVALIDATION_MESSAGE = [b"invalidate", "invalidate"]
+
 
 class _RESP3Parser(_RESPBase):
     """RESP3 protocol implementation"""
 
     def __init__(self, socket_read_size):
         super().__init__(socket_read_size)
-        self.push_handler_func = self.handle_push_response
+        self.pubsub_push_handler_func = self.handle_pubsub_push_response
+        self.invalidations_push_handler_func = None
 
-    def handle_push_response(self, response):
+    def handle_pubsub_push_response(self, response):
         logger = getLogger("push_response")
         logger.info("Push response: " + str(response))
         return response
@@ -96,8 +99,9 @@ def _read_response(self, disable_decoding=False, push_request=False):
                 pass
         # map response
         elif byte == b"%":
-            # we use this approach and not dict comprehension here
-            # because this dict comprehension fails in python 3.7
+            # We cannot use a dict-comprehension to parse stream.
+            # Evaluation order of key:val expression in dict comprehension only
+            # became defined to be left-right in version 3.8
             resp_dict = {}
             for _ in range(int(response)):
                 key = self._read_response(disable_decoding=disable_decoding)
@@ -113,30 +117,40 @@ def _read_response(self, disable_decoding=False, push_request=False):
                 )
                 for _ in range(int(response))
             ]
-            res = self.push_handler_func(response)
-            if not push_request:
-                return self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
-                )
-            else:
-                return res
+            self.handle_push_response(response, disable_decoding, push_request)
         else:
             raise InvalidResponse(f"Protocol Error: {raw!r}")
 
         if isinstance(response, bytes) and disable_decoding is False:
             response = self.encoder.decode(response)
         return response
 
-    def set_push_handler(self, push_handler_func):
-        self.push_handler_func = push_handler_func
+    def handle_push_response(self, response, disable_decoding, push_request):
+        if response[0] in _INVALIDATION_MESSAGE:
+            res = self.invalidation_push_handler_func(response)
+        else:
+            res = self.pubsub_push_handler_func(response)
+        if not push_request:
+            return self._read_response(
+                disable_decoding=disable_decoding, push_request=push_request
+            )
+        else:
+            return res
+
+    def set_pubsub_push_handler(self, pubsub_push_handler_func):
+        self.pubsub_push_handler_func = pubsub_push_handler_func
+
+    def set_invalidation_push_handler(self, invalidations_push_handler_func):
+        self.invalidation_push_handler_func = invalidations_push_handler_func
 
 
 class _AsyncRESP3Parser(_AsyncRESPBase):
     def __init__(self, socket_read_size):
         super().__init__(socket_read_size)
-        self.push_handler_func = self.handle_push_response
+        self.pubsub_push_handler_func = self.handle_pubsub_push_response
+        self.invalidations_push_handler_func = None
 
-    def handle_push_response(self, response):
+    def handle_pubsub_push_response(self, response):
         logger = getLogger("push_response")
         logger.info("Push response: " + str(response))
         return response
@@ -225,12 +239,16 @@ async def _read_response(
                 pass
         # map response
         elif byte == b"%":
-            response = {
-                (await self._read_response(disable_decoding=disable_decoding)): (
-                    await self._read_response(disable_decoding=disable_decoding)
+            # We cannot use a dict-comprehension to parse stream.
+            # Evaluation order of key:val expression in dict comprehension only
+            # became defined to be left-right in version 3.8
+            resp_dict = {}
+            for _ in range(int(response)):
+                key = await self._read_response(disable_decoding=disable_decoding)
+                resp_dict[key] = await self._read_response(
+                    disable_decoding=disable_decoding, push_request=push_request
                 )
-                for _ in range(int(response))
-            }
+            response = resp_dict
         # push response
         elif byte == b">":
             response = [
@@ -241,19 +259,28 @@ async def _read_response(
                 )
                 for _ in range(int(response))
             ]
-            res = self.push_handler_func(response)
-            if not push_request:
-                return await self._read_response(
-                    disable_decoding=disable_decoding, push_request=push_request
-                )
-            else:
-                return res
+            await self.handle_push_response(response, disable_decoding, push_request)
         else:
             raise InvalidResponse(f"Protocol Error: {raw!r}")
 
         if isinstance(response, bytes) and disable_decoding is False:
             response = self.encoder.decode(response)
         return response
 
-    def set_push_handler(self, push_handler_func):
-        self.push_handler_func = push_handler_func
+    async def handle_push_response(self, response, disable_decoding, push_request):
+        if response[0] in _INVALIDATION_MESSAGE:
+            res = self.invalidation_push_handler_func(response)
+        else:
+            res = self.pubsub_push_handler_func(response)
+        if not push_request:
+            return await self._read_response(
+                disable_decoding=disable_decoding, push_request=push_request
+            )
+        else:
+            return res
+
+    def set_pubsub_push_handler(self, pubsub_push_handler_func):
+        self.pubsub_push_handler_func = pubsub_push_handler_func
+
+    def set_invalidation_push_handler(self, invalidations_push_handler_func):
+        self.invalidation_push_handler_func = invalidations_push_handler_func