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

Author: petyaslavova

redis/_parsers/__init__.py

@@ -1,4 +1,9 @@
-from .base import AsyncPushNotificationsParser, BaseParser, PushNotificationsParser, _AsyncRESPBase
+from .base import (
+    AsyncPushNotificationsParser,
+    BaseParser,
+    PushNotificationsParser,
+    _AsyncRESPBase,
+)
 from .commands import AsyncCommandsParser, CommandsParser
 from .encoders import Encoder
 from .hiredis import _AsyncHiredisParser, _HiredisParser

redis/_parsers/hiredis.py

@@ -14,9 +14,9 @@
 from ..utils import HIREDIS_AVAILABLE
 from .base import (
     AsyncBaseParser,
+    AsyncPushNotificationsParser,
     BaseParser,
     PushNotificationsParser,
-    AsyncPushNotificationsParser,
 )
 from .socket import (
     NONBLOCKING_EXCEPTION_ERROR_NUMBERS,
@@ -137,6 +137,16 @@ def read_response(self, disable_decoding=False, push_request=False):
         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:
@@ -174,7 +184,7 @@ def read_response(self, disable_decoding=False, push_request=False):
         return response
 
 
-class _AsyncHiredisParser(AsyncBaseParser):
+class _AsyncHiredisParser(AsyncBaseParser, AsyncPushNotificationsParser):
     """Async implementation of parser class for connections using Hiredis"""
 
     __slots__ = ("_reader",)

tests/test_asyncio/test_search.py

@@ -7,6 +7,7 @@
 import numpy as np
 import pytest
 import pytest_asyncio
+from redis import ResponseError
 import redis.asyncio as redis
 import redis.commands.search.aggregation as aggregations
 import redis.commands.search.reducers as reducers
@@ -60,6 +61,10 @@ async def waitForIndex(env, idx, timeout=None):
                     break
             except ValueError:
                 break
+        except ResponseError:
+            # index doesn't exist yet
+            # continue to sleep and try again
+            pass
 
         await asyncio.sleep(delay)
         if timeout is not None:

tests/test_cache.py

@@ -13,7 +13,6 @@
     EvictionPolicyType,
     LRUPolicy,
 )
-from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import _get_client, skip_if_resp_version, skip_if_server_version_lt
 
 

tests/test_connection_pool.py

@@ -9,7 +9,7 @@
 import redis
 from redis.cache import CacheConfig
 from redis.connection import CacheProxyConnection, Connection, to_bool
-from redis.utils import HIREDIS_AVAILABLE, SSL_AVAILABLE
+from redis.utils import SSL_AVAILABLE
 
 from .conftest import (
     _get_client,

tests/test_pubsub.py

@@ -10,7 +10,6 @@
 import pytest
 import redis
 from redis.exceptions import ConnectionError
-from redis.utils import HIREDIS_AVAILABLE
 
 from .conftest import (
     _get_client,

tests/test_search.py

@@ -6,6 +6,7 @@
 
 import numpy as np
 import pytest
+from redis import ResponseError
 import redis
 import redis.commands.search
 import redis.commands.search.aggregation as aggregations
@@ -47,8 +48,8 @@
 def waitForIndex(env, idx, timeout=None):
     delay = 0.1
     while True:
-        res = env.execute_command("FT.INFO", idx)
         try:
+            res = env.execute_command("FT.INFO", idx)
             if int(res[res.index("indexing") + 1]) == 0:
                 break
         except ValueError:
@@ -59,6 +60,10 @@ def waitForIndex(env, idx, timeout=None):
                     break
             except ValueError:
                 break
+        except ResponseError:
+            # index doesn't exist yet
+            # continue to sleep and try again
+            pass
 
         time.sleep(delay)
         if timeout is not None:
@@ -1909,6 +1914,8 @@ def test_binary_and_text_fields(client):
         ),
     )
 
+    waitForIndex(client, index_name)
+
     query = (
         Query("*")
         .return_field("vector_emb", decode_field=False)