Support RESP3 with hiredis-py parser (#3648)

* Add minimalistic implementation for Push notifications parsing with hiredis-py

* Fixing linters(almost all errors). Fixing hiredis parser for invalidation messages when using connection pool.

* Fixing last linter errors. Fixes needed for pubsub message handling with hiredis.

* Fix handle_pubsub_push_response

* Update redis/utils.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Fix formatting

* Some good sleep solves a lot of problems. Isn't it?

* Enable verbose output from pytest to check why tests are skipped

* Revert

* Bump hiredis version to 3.2.0

* Bump hiredis-py version in CI

---------

Co-authored-by: Petya Slavova <petya.slavova@redis.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

.github/workflows/integration.yaml

@@ -125,7 +125,7 @@ jobs:
         redis-version: [ '${{ needs.redis_version.outputs.CURRENT }}' ]
         python-version: [ '3.8', '3.13']
         parser-backend: [ 'hiredis' ]
-        hiredis-version: [ '>=3.0.0', '<3.0.0' ]
+        hiredis-version: [ '>=3.2.0', '<3.0.0' ]
         event-loop: [ 'asyncio' ]
     env:
       ACTIONS_ALLOW_UNSECURE_COMMANDS: true

pyproject.toml

@@ -33,7 +33,7 @@ dependencies = ['async-timeout>=4.0.3; python_full_version<"3.11.3"']
 
 [project.optional-dependencies]
 hiredis = [
-    "hiredis>=3.0.0",
+    "hiredis>=3.2.0",
 ]
 ocsp = [
     "cryptography>=36.0.1",

redis/_parsers/__init__.py

@@ -1,4 +1,9 @@
-from .base import BaseParser, _AsyncRESPBase
+from .base import (
+    AsyncPushNotificationsParser,
+    BaseParser,
+    PushNotificationsParser,
+    _AsyncRESPBase,
+)
 from .commands import AsyncCommandsParser, CommandsParser
 from .encoders import Encoder
 from .hiredis import _AsyncHiredisParser, _HiredisParser
@@ -11,10 +16,12 @@
     "_AsyncRESPBase",
     "_AsyncRESP2Parser",
     "_AsyncRESP3Parser",
+    "AsyncPushNotificationsParser",
     "CommandsParser",
     "Encoder",
     "BaseParser",
     "_HiredisParser",
     "_RESP2Parser",
     "_RESP3Parser",
+    "PushNotificationsParser",
 ]

redis/_parsers/base.py

@@ -1,7 +1,7 @@
 import sys
 from abc import ABC
 from asyncio import IncompleteReadError, StreamReader, TimeoutError
-from typing import List, Optional, Union
+from typing import Callable, List, Optional, Protocol, Union
 
 if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
     from asyncio import timeout as async_timeout
@@ -158,6 +158,58 @@ async def read_response(
         raise NotImplementedError()
 
 
+_INVALIDATION_MESSAGE = [b"invalidate", "invalidate"]
+
+
+class PushNotificationsParser(Protocol):
+    """Protocol defining RESP3-specific parsing functionality"""
+
+    pubsub_push_handler_func: Callable
+    invalidation_push_handler_func: Optional[Callable] = None
+
+    def handle_pubsub_push_response(self, response):
+        """Handle pubsub push responses"""
+        raise NotImplementedError()
+
+    def handle_push_response(self, response, **kwargs):
+        if response[0] not in _INVALIDATION_MESSAGE:
+            return self.pubsub_push_handler_func(response)
+        if self.invalidation_push_handler_func:
+            return self.invalidation_push_handler_func(response)
+
+    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, invalidation_push_handler_func):
+        self.invalidation_push_handler_func = invalidation_push_handler_func
+
+
+class AsyncPushNotificationsParser(Protocol):
+    """Protocol defining async RESP3-specific parsing functionality"""
+
+    pubsub_push_handler_func: Callable
+    invalidation_push_handler_func: Optional[Callable] = None
+
+    async def handle_pubsub_push_response(self, response):
+        """Handle pubsub push responses asynchronously"""
+        raise NotImplementedError()
+
+    async def handle_push_response(self, response, **kwargs):
+        """Handle push responses asynchronously"""
+        if response[0] not in _INVALIDATION_MESSAGE:
+            return await self.pubsub_push_handler_func(response)
+        if self.invalidation_push_handler_func:
+            return await self.invalidation_push_handler_func(response)
+
+    def set_pubsub_push_handler(self, pubsub_push_handler_func):
+        """Set the pubsub push handler function"""
+        self.pubsub_push_handler_func = pubsub_push_handler_func
+
+    def set_invalidation_push_handler(self, invalidation_push_handler_func):
+        """Set the invalidation push handler function"""
+        self.invalidation_push_handler_func = invalidation_push_handler_func
+
+
 class _AsyncRESPBase(AsyncBaseParser):
     """Base class for async resp parsing"""
 

redis/_parsers/hiredis.py

@@ -1,6 +1,7 @@
 import asyncio
 import socket
 import sys
+from logging import getLogger
 from typing import Callable, List, Optional, TypedDict, Union
 
 if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
@@ -11,7 +12,12 @@
 from ..exceptions import ConnectionError, InvalidResponse, RedisError
 from ..typing import EncodableT
 from ..utils import HIREDIS_AVAILABLE
-from .base import AsyncBaseParser, BaseParser
+from .base import (
+    AsyncBaseParser,
+    AsyncPushNotificationsParser,
+    BaseParser,
+    PushNotificationsParser,
+)
 from .socket import (
     NONBLOCKING_EXCEPTION_ERROR_NUMBERS,
     NONBLOCKING_EXCEPTIONS,
@@ -32,21 +38,29 @@ class _HiredisReaderArgs(TypedDict, total=False):
     errors: Optional[str]
 
 
-class _HiredisParser(BaseParser):
+class _HiredisParser(BaseParser, PushNotificationsParser):
     "Parser class for connections using Hiredis"
 
     def __init__(self, socket_read_size):
         if not HIREDIS_AVAILABLE:
             raise RedisError("Hiredis is not installed")
         self.socket_read_size = socket_read_size
         self._buffer = bytearray(socket_read_size)
+        self.pubsub_push_handler_func = self.handle_pubsub_push_response
+        self.invalidation_push_handler_func = None
+        self._hiredis_PushNotificationType = None
 
     def __del__(self):
         try:
             self.on_disconnect()
         except Exception:
             pass
 
+    def handle_pubsub_push_response(self, response):
+        logger = getLogger("push_response")
+        logger.debug("Push response: " + str(response))
+        return response
+
     def on_connect(self, connection, **kwargs):
         import hiredis
 
@@ -64,6 +78,12 @@ def on_connect(self, connection, **kwargs):
         self._reader = hiredis.Reader(**kwargs)
         self._next_response = NOT_ENOUGH_DATA
 
+        try:
+            self._hiredis_PushNotificationType = hiredis.PushNotification
+        except AttributeError:
+            # hiredis < 3.2
+            self._hiredis_PushNotificationType = None
+
     def on_disconnect(self):
         self._sock = None
         self._reader = None
@@ -109,14 +129,24 @@ def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True):
             if custom_timeout:
                 sock.settimeout(self._socket_timeout)
 
-    def read_response(self, disable_decoding=False):
+    def read_response(self, disable_decoding=False, push_request=False):
         if not self._reader:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
 
         # _next_response might be cached from a can_read() call
         if self._next_response is not NOT_ENOUGH_DATA:
             response = self._next_response
             self._next_response = NOT_ENOUGH_DATA
+            if self._hiredis_PushNotificationType is not None and isinstance(
+                response, self._hiredis_PushNotificationType
+            ):
+                response = self.handle_push_response(response)
+                if not push_request:
+                    return self.read_response(
+                        disable_decoding=disable_decoding, push_request=push_request
+                    )
+                else:
+                    return response
             return response
 
         if disable_decoding:
@@ -135,6 +165,16 @@ def read_response(self, disable_decoding=False):
         # happened
         if isinstance(response, ConnectionError):
             raise response
+        elif self._hiredis_PushNotificationType is not None and isinstance(
+            response, self._hiredis_PushNotificationType
+        ):
+            response = self.handle_push_response(response)
+            if not push_request:
+                return self.read_response(
+                    disable_decoding=disable_decoding, push_request=push_request
+                )
+            else:
+                return response
         elif (
             isinstance(response, list)
             and response
@@ -144,7 +184,7 @@ def read_response(self, disable_decoding=False):
         return response
 
 
-class _AsyncHiredisParser(AsyncBaseParser):
+class _AsyncHiredisParser(AsyncBaseParser, AsyncPushNotificationsParser):
     """Async implementation of parser class for connections using Hiredis"""
 
     __slots__ = ("_reader",)
@@ -154,6 +194,14 @@ def __init__(self, socket_read_size: int):
             raise RedisError("Hiredis is not available.")
         super().__init__(socket_read_size=socket_read_size)
         self._reader = None
+        self.pubsub_push_handler_func = self.handle_pubsub_push_response
+        self.invalidation_push_handler_func = None
+        self._hiredis_PushNotificationType = None
+
+    async def handle_pubsub_push_response(self, response):
+        logger = getLogger("push_response")
+        logger.debug("Push response: " + str(response))
+        return response
 
     def on_connect(self, connection):
         import hiredis
@@ -171,6 +219,14 @@ def on_connect(self, connection):
         self._reader = hiredis.Reader(**kwargs)
         self._connected = True
 
+        try:
+            self._hiredis_PushNotificationType = getattr(
+                hiredis, "PushNotification", None
+            )
+        except AttributeError:
+            # hiredis < 3.2
+            self._hiredis_PushNotificationType = None
+
     def on_disconnect(self):
         self._connected = False
 
@@ -195,7 +251,7 @@ async def read_from_socket(self):
         return True
 
     async def read_response(
-        self, disable_decoding: bool = False
+        self, disable_decoding: bool = False, push_request: bool = False
     ) -> Union[EncodableT, List[EncodableT]]:
         # If `on_disconnect()` has been called, prohibit any more reads
         # even if they could happen because data might be present.
@@ -207,6 +263,7 @@ async def read_response(
             response = self._reader.gets(False)
         else:
             response = self._reader.gets()
+
         while response is NOT_ENOUGH_DATA:
             await self.read_from_socket()
             if disable_decoding:
@@ -219,6 +276,16 @@ async def read_response(
         # happened
         if isinstance(response, ConnectionError):
             raise response
+        elif self._hiredis_PushNotificationType is not None and isinstance(
+            response, self._hiredis_PushNotificationType
+        ):
+            response = await self.handle_push_response(response)
+            if not push_request:
+                return await self.read_response(
+                    disable_decoding=disable_decoding, push_request=push_request
+                )
+            else:
+                return response
         elif (
             isinstance(response, list)
             and response

redis/_parsers/resp3.py

@@ -3,13 +3,16 @@
 
 from ..exceptions import ConnectionError, InvalidResponse, ResponseError
 from ..typing import EncodableT
-from .base import _AsyncRESPBase, _RESPBase
+from .base import (
+    AsyncPushNotificationsParser,
+    PushNotificationsParser,
+    _AsyncRESPBase,
+    _RESPBase,
+)
 from .socket import SERVER_CLOSED_CONNECTION_ERROR
 
-_INVALIDATION_MESSAGE = [b"invalidate", "invalidate"]
 
-
-class _RESP3Parser(_RESPBase):
+class _RESP3Parser(_RESPBase, PushNotificationsParser):
     """RESP3 protocol implementation"""
 
     def __init__(self, socket_read_size):
@@ -113,9 +116,7 @@ def _read_response(self, disable_decoding=False, push_request=False):
                 )
                 for _ in range(int(response))
             ]
-            response = self.handle_push_response(
-                response, disable_decoding, push_request
-            )
+            response = self.handle_push_response(response)
             if not push_request:
                 return self._read_response(
                     disable_decoding=disable_decoding, push_request=push_request
@@ -129,20 +130,8 @@ def _read_response(self, disable_decoding=False, push_request=False):
             response = self.encoder.decode(response)
         return response
 
-    def handle_push_response(self, response, disable_decoding, push_request):
-        if response[0] not in _INVALIDATION_MESSAGE:
-            return self.pubsub_push_handler_func(response)
-        if self.invalidation_push_handler_func:
-            return self.invalidation_push_handler_func(response)
-
-    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, invalidation_push_handler_func):
-        self.invalidation_push_handler_func = invalidation_push_handler_func
-
 
-class _AsyncRESP3Parser(_AsyncRESPBase):
+class _AsyncRESP3Parser(_AsyncRESPBase, AsyncPushNotificationsParser):
     def __init__(self, socket_read_size):
         super().__init__(socket_read_size)
         self.pubsub_push_handler_func = self.handle_pubsub_push_response
@@ -253,9 +242,7 @@ async def _read_response(
                 )
                 for _ in range(int(response))
             ]
-            response = await self.handle_push_response(
-                response, disable_decoding, push_request
-            )
+            response = await self.handle_push_response(response)
             if not push_request:
                 return await self._read_response(
                     disable_decoding=disable_decoding, push_request=push_request
@@ -268,15 +255,3 @@ async def _read_response(
         if isinstance(response, bytes) and disable_decoding is False:
             response = self.encoder.decode(response)
         return response
-
-    async def handle_push_response(self, response, disable_decoding, push_request):
-        if response[0] not in _INVALIDATION_MESSAGE:
-            return await self.pubsub_push_handler_func(response)
-        if self.invalidation_push_handler_func:
-            return await self.invalidation_push_handler_func(response)
-
-    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, invalidation_push_handler_func):
-        self.invalidation_push_handler_func = invalidation_push_handler_func

redis/asyncio/client.py

@@ -70,7 +70,6 @@
 )
 from redis.typing import ChannelT, EncodableT, KeyT
 from redis.utils import (
-    HIREDIS_AVAILABLE,
     SSL_AVAILABLE,
     _set_info_logger,
     deprecated_args,
@@ -938,7 +937,7 @@ async def connect(self):
             self.connection.register_connect_callback(self.on_connect)
         else:
             await self.connection.connect()
-        if self.push_handler_func is not None and not HIREDIS_AVAILABLE:
+        if self.push_handler_func is not None:
             self.connection._parser.set_pubsub_push_handler(self.push_handler_func)
 
         self._event_dispatcher.dispatch(