Reorganizing the parsers code, and add support for RESP3 (#2574)

* Reorganizing the parsers code

* fix build package

* fix imports

* fix flake8

* add resp to Connection class

* core commands

* python resp3 parser

* pipeline

* async resp3 parser

* some asymc tests

* resp3 parser for async cluster

* async commands tests

* linters

* linters

* linters

* fix ModuleNotFoundError

* fix tests

* fix assert_resp_response_in

* fix command_getkeys in cluster

* fail-fast false

* version

---------

Co-authored-by: Chayim I. Kirshen <[email protected]>

Author: dvora-h

.github/workflows/integration.yaml

@@ -51,6 +51,7 @@ jobs:
      timeout-minutes: 30
      strategy:
        max-parallel: 15
+       fail-fast: false
        matrix:
          python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
          test-type: ['standalone', 'cluster']
@@ -108,6 +109,7 @@ jobs:
     name: Install package from commit hash
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
     steps:

benchmarks/socket_read_size.py

@@ -1,12 +1,12 @@
 from base import Benchmark
 
-from redis.connection import HiredisParser, PythonParser
+from redis.connection import PythonParser, _HiredisParser
 
 
 class SocketReadBenchmark(Benchmark):
 
     ARGUMENTS = (
-        {"name": "parser", "values": [PythonParser, HiredisParser]},
+        {"name": "parser", "values": [PythonParser, _HiredisParser]},
         {
             "name": "value_size",
             "values": [10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000],

redis/asyncio/__init__.py

@@ -7,7 +7,6 @@
     SSLConnection,
     UnixDomainSocketConnection,
 )
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.sentinel import (
     Sentinel,
     SentinelConnectionPool,
@@ -38,7 +37,6 @@
     "BlockingConnectionPool",
     "BusyLoadingError",
     "ChildDeadlockedError",
-    "CommandsParser",
     "Connection",
     "ConnectionError",
     "ConnectionPool",

redis/asyncio/client.py

@@ -253,6 +253,9 @@ def __init__(
 
         self.response_callbacks = CaseInsensitiveDict(self.__class__.RESPONSE_CALLBACKS)
 
+        if self.connection_pool.connection_kwargs.get("protocol") == "3":
+            self.response_callbacks.update(self.__class__.RESP3_RESPONSE_CALLBACKS)
+
         # If using a single connection client, we need to lock creation-of and use-of
         # the client in order to avoid race conditions such as using asyncio.gather
         # on a set of redis commands

redis/asyncio/cluster.py

@@ -17,15 +17,8 @@
 )
 
 from redis.asyncio.client import ResponseCallbackT
-from redis.asyncio.connection import (
-    Connection,
-    DefaultParser,
-    Encoder,
-    SSLConnection,
-    parse_url,
-)
+from redis.asyncio.connection import Connection, DefaultParser, SSLConnection, parse_url
 from redis.asyncio.lock import Lock
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.retry import Retry
 from redis.backoff import default_backoff
 from redis.client import EMPTY_RESPONSE, NEVER_DECODE, AbstractRedis
@@ -60,6 +53,7 @@
     TimeoutError,
     TryAgainError,
 )
+from redis.parsers import AsyncCommandsParser, Encoder
 from redis.typing import AnyKeyT, EncodableT, KeyT
 from redis.utils import dict_merge, safe_str, str_if_bytes
 
@@ -250,6 +244,7 @@ def __init__(
         ssl_certfile: Optional[str] = None,
         ssl_check_hostname: bool = False,
         ssl_keyfile: Optional[str] = None,
+        protocol: Optional[int] = 2,
     ) -> None:
         if db:
             raise RedisClusterException(
@@ -290,6 +285,7 @@ def __init__(
             "socket_keepalive_options": socket_keepalive_options,
             "socket_timeout": socket_timeout,
             "retry": retry,
+            "protocol": protocol,
         }
 
         if ssl:
@@ -344,7 +340,7 @@ def __init__(
         self.cluster_error_retry_attempts = cluster_error_retry_attempts
         self.connection_error_retry_attempts = connection_error_retry_attempts
         self.reinitialize_counter = 0
-        self.commands_parser = CommandsParser()
+        self.commands_parser = AsyncCommandsParser()
         self.node_flags = self.__class__.NODE_FLAGS.copy()
         self.command_flags = self.__class__.COMMAND_FLAGS.copy()
         self.response_callbacks = kwargs["response_callbacks"]