Reorganizing the code, in preparation for RESP3

27 files changed, 892 insertions, 895 deletions

diff --git a/redis/asyncio/__init__.py b/redis/asyncio/__init__.py
index bf90dde..7b95083 100644
--- a/redis/asyncio/__init__.py
+++ b/redis/asyncio/__init__.py
@@ -7,7 +7,6 @@ from redis.asyncio.connection import (
     SSLConnection,
     UnixDomainSocketConnection,
 )
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.sentinel import (
     Sentinel,
     SentinelConnectionPool,
@@ -38,7 +37,6 @@ __all__ = [
     "BlockingConnectionPool",
     "BusyLoadingError",
     "ChildDeadlockedError",
-    "CommandsParser",
     "Connection",
     "ConnectionError",
     "ConnectionPool",
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index d5a38b2..a1762db 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -17,15 +17,8 @@ from typing import (
 )
 
 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 @@ from redis.exceptions import (
     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
 
@@ -344,7 +338,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
         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"]
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 4f19153..6275632 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -33,16 +33,9 @@ from redis.credentials import CredentialProvider, UsernamePasswordCredentialProv
 from redis.exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
-    BusyLoadingError,
     ChildDeadlockedError,
     ConnectionError,
     DataError,
-    ExecAbortError,
-    InvalidResponse,
-    ModuleError,
-    NoPermissionError,
-    NoScriptError,
-    ReadOnlyError,
     RedisError,
     ResponseError,
     TimeoutError,
@@ -50,9 +43,7 @@ from redis.exceptions import (
 from redis.typing import EncodableT, EncodedT
 from redis.utils import HIREDIS_AVAILABLE, str_if_bytes
 
-hiredis = None
-if HIREDIS_AVAILABLE:
-    import hiredis
+from ..parsers import AsyncHiredisParser, AsyncRESP2Parser, BaseParser, Encoder
 
 SYM_STAR = b"*"
 SYM_DOLLAR = b"$"
@@ -60,8 +51,6 @@ SYM_CRLF = b"\r\n"
 SYM_LF = b"\n"
 SYM_EMPTY = b""
 
-SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server."
-
 
 class _Sentinel(enum.Enum):
     sentinel = object()
@@ -87,311 +76,11 @@ NO_AUTH_SET_ERROR = {
 }
 
 
-class _HiredisReaderArgs(TypedDict, total=False):
-    protocolError: Callable[[str], Exception]
-    replyError: Callable[[str], Exception]
-    encoding: Optional[str]
-    errors: Optional[str]
-
-
-class Encoder:
-    """Encode strings to bytes-like and decode bytes-like to strings"""
-
-    __slots__ = "encoding", "encoding_errors", "decode_responses"
-
-    def __init__(self, encoding: str, encoding_errors: str, decode_responses: bool):
-        self.encoding = encoding
-        self.encoding_errors = encoding_errors
-        self.decode_responses = decode_responses
-
-    def encode(self, value: EncodableT) -> EncodedT:
-        """Return a bytestring or bytes-like representation of the value"""
-        if isinstance(value, str):
-            return value.encode(self.encoding, self.encoding_errors)
-        if isinstance(value, (bytes, memoryview)):
-            return value
-        if isinstance(value, (int, float)):
-            if isinstance(value, bool):
-                # special case bool since it is a subclass of int
-                raise DataError(
-                    "Invalid input of type: 'bool'. "
-                    "Convert to a bytes, string, int or float first."
-                )
-            return repr(value).encode()
-        # a value we don't know how to deal with. throw an error
-        typename = value.__class__.__name__
-        raise DataError(
-            f"Invalid input of type: {typename!r}. "
-            "Convert to a bytes, string, int or float first."
-        )
-
-    def decode(self, value: EncodableT, force=False) -> EncodableT:
-        """Return a unicode string from the bytes-like representation"""
-        if self.decode_responses or force:
-            if isinstance(value, bytes):
-                return value.decode(self.encoding, self.encoding_errors)
-            if isinstance(value, memoryview):
-                return value.tobytes().decode(self.encoding, self.encoding_errors)
-        return value
-
-
-ExceptionMappingT = Mapping[str, Union[Type[Exception], Mapping[str, Type[Exception]]]]
-
-
-class BaseParser:
-    """Plain Python parsing class"""
-
-    __slots__ = "_stream", "_read_size"
-
-    EXCEPTION_CLASSES: ExceptionMappingT = {
-        "ERR": {
-            "max number of clients reached": ConnectionError,
-            "Client sent AUTH, but no password is set": AuthenticationError,
-            "invalid password": AuthenticationError,
-            # some Redis server versions report invalid command syntax
-            # in lowercase
-            "wrong number of arguments for 'auth' command": AuthenticationWrongNumberOfArgsError,  # noqa: E501
-            # some Redis server versions report invalid command syntax
-            # in uppercase
-            "wrong number of arguments for 'AUTH' command": AuthenticationWrongNumberOfArgsError,  # noqa: E501
-            MODULE_LOAD_ERROR: ModuleError,
-            MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError,
-            NO_SUCH_MODULE_ERROR: ModuleError,
-            MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,
-            **NO_AUTH_SET_ERROR,
-        },
-        "WRONGPASS": AuthenticationError,
-        "EXECABORT": ExecAbortError,
-        "LOADING": BusyLoadingError,
-        "NOSCRIPT": NoScriptError,
-        "READONLY": ReadOnlyError,
-        "NOAUTH": AuthenticationError,
-        "NOPERM": NoPermissionError,
-    }
-
-    def __init__(self, socket_read_size: int):
-        self._stream: Optional[asyncio.StreamReader] = None
-        self._read_size = socket_read_size
-
-    def __del__(self):
-        try:
-            self.on_disconnect()
-        except Exception:
-            pass
-
-    def parse_error(self, response: str) -> ResponseError:
-        """Parse an error response"""
-        error_code = response.split(" ")[0]
-        if error_code in self.EXCEPTION_CLASSES:
-            response = response[len(error_code) + 1 :]
-            exception_class = self.EXCEPTION_CLASSES[error_code]
-            if isinstance(exception_class, dict):
-                exception_class = exception_class.get(response, ResponseError)
-            return exception_class(response)
-        return ResponseError(response)
-
-    def on_disconnect(self):
-        raise NotImplementedError()
-
-    def on_connect(self, connection: "Connection"):
-        raise NotImplementedError()
-
-    async def can_read_destructive(self) -> bool:
-        raise NotImplementedError()
-
-    async def read_response(
-        self, disable_decoding: bool = False
-    ) -> Union[EncodableT, ResponseError, None, List[EncodableT]]:
-        raise NotImplementedError()
-
-
-class PythonParser(BaseParser):
-    """Plain Python parsing class"""
-
-    __slots__ = BaseParser.__slots__ + ("encoder",)
-
-    def __init__(self, socket_read_size: int):
-        super().__init__(socket_read_size)
-        self.encoder: Optional[Encoder] = None
-
-    def on_connect(self, connection: "Connection"):
-        """Called when the stream connects"""
-        self._stream = connection._reader
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-
-        self.encoder = connection.encoder
-
-    def on_disconnect(self):
-        """Called when the stream disconnects"""
-        if self._stream is not None:
-            self._stream = None
-        self.encoder = None
-
-    async def can_read_destructive(self) -> bool:
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-        try:
-            async with async_timeout.timeout(0):
-                return await self._stream.read(1)
-        except asyncio.TimeoutError:
-            return False
-
-    async def read_response(
-        self, disable_decoding: bool = False
-    ) -> Union[EncodableT, ResponseError, None]:
-        if not self._stream or not self.encoder:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        raw = await self._readline()
-        if not raw:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        response: Any
-        byte, response = raw[:1], raw[1:]
-
-        if byte not in (b"-", b"+", b":", b"$", b"*"):
-            raise InvalidResponse(f"Protocol Error: {raw!r}")
-
-        # server returned an error
-        if byte == b"-":
-            response = response.decode("utf-8", errors="replace")
-            error = self.parse_error(response)
-            # if the error is a ConnectionError, raise immediately so the user
-            # is notified
-            if isinstance(error, ConnectionError):
-                raise error
-            # otherwise, we're dealing with a ResponseError that might belong
-            # inside a pipeline response. the connection's read_response()
-            # and/or the pipeline's execute() will raise this error if
-            # necessary, so just return the exception instance here.
-            return error
-        # single value
-        elif byte == b"+":
-            pass
-        # int value
-        elif byte == b":":
-            response = int(response)
-        # bulk response
-        elif byte == b"$":
-            length = int(response)
-            if length == -1:
-                return None
-            response = await self._read(length)
-        # multi-bulk response
-        elif byte == b"*":
-            length = int(response)
-            if length == -1:
-                return None
-            response = [
-                (await self.read_response(disable_decoding)) for _ in range(length)
-            ]
-        if isinstance(response, bytes) and disable_decoding is False:
-            response = self.encoder.decode(response)
-        return response
-
-    async def _read(self, length: int) -> bytes:
-        """
-        Read `length` bytes of data.  These are assumed to be followed
-        by a '\r\n' terminator which is subsequently discarded.
-        """
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-        try:
-            data = await self._stream.readexactly(length + 2)
-        except asyncio.IncompleteReadError as error:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error
-        return data[:-2]
-
-    async def _readline(self) -> bytes:
-        """
-        read an unknown number of bytes up to the next '\r\n'
-        line separator, which is discarded.
-        """
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-        data = await self._stream.readline()
-        if not data.endswith(b"\r\n"):
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        return data[:-2]
-
-
-class HiredisParser(BaseParser):
-    """Parser class for connections using Hiredis"""
-
-    __slots__ = BaseParser.__slots__ + ("_reader",)
-
-    def __init__(self, socket_read_size: int):
-        if not HIREDIS_AVAILABLE:
-            raise RedisError("Hiredis is not available.")
-        super().__init__(socket_read_size=socket_read_size)
-        self._reader: Optional[hiredis.Reader] = None
-
-    def on_connect(self, connection: "Connection"):
-        self._stream = connection._reader
-        kwargs: _HiredisReaderArgs = {
-            "protocolError": InvalidResponse,
-            "replyError": self.parse_error,
-        }
-        if connection.encoder.decode_responses:
-            kwargs["encoding"] = connection.encoder.encoding
-            kwargs["errors"] = connection.encoder.encoding_errors
-
-        self._reader = hiredis.Reader(**kwargs)
-
-    def on_disconnect(self):
-        self._stream = None
-        self._reader = None
-
-    async def can_read_destructive(self):
-        if not self._stream or not self._reader:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        if self._reader.gets():
-            return True
-        try:
-            async with async_timeout.timeout(0):
-                return await self.read_from_socket()
-        except asyncio.TimeoutError:
-            return False
-
-    async def read_from_socket(self):
-        buffer = await self._stream.read(self._read_size)
-        if not buffer or not isinstance(buffer, bytes):
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
-        self._reader.feed(buffer)
-        # data was read from the socket and added to the buffer.
-        # return True to indicate that data was read.
-        return True
-
-    async def read_response(
-        self, disable_decoding: bool = False
-    ) -> Union[EncodableT, List[EncodableT]]:
-        if not self._stream or not self._reader:
-            self.on_disconnect()
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
-
-        response = self._reader.gets()
-        while response is False:
-            await self.read_from_socket()
-            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
-        # happened
-        if isinstance(response, ConnectionError):
-            raise response
-        elif (
-            isinstance(response, list)
-            and response
-            and isinstance(response[0], ConnectionError)
-        ):
-            raise response[0]
-        return response
-
-
-DefaultParser: Type[Union[PythonParser, HiredisParser]]
+DefaultParser: Type[Union[AsyncRESP2Parser, AsyncHiredisParser]]
 if HIREDIS_AVAILABLE:
-    DefaultParser = HiredisParser
+    DefaultParser = AsyncHiredisParser
 else:
-    DefaultParser = PythonParser
+    DefaultParser = AsyncRESP2Parser
 
 
 class ConnectCallbackProtocol(Protocol):
diff --git a/redis/asyncio/parser.py b/redis/asyncio/parser.py
deleted file mode 100644
index 5faf8f8..0000000
--- a/redis/asyncio/parser.py
+++ /dev/null
@@ -1,94 +0,0 @@
-from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
-
-from redis.exceptions import RedisError, ResponseError
-
-if TYPE_CHECKING:
-    from redis.asyncio.cluster import ClusterNode
-
-
-class CommandsParser:
-    """
-    Parses Redis commands to get command keys.
-
-    COMMAND output is used to determine key locations.
-    Commands that do not have a predefined key location are flagged with 'movablekeys',
-    and these commands' keys are determined by the command 'COMMAND GETKEYS'.
-
-    NOTE: Due to a bug in redis<7.0, this does not work properly
-    for EVAL or EVALSHA when the `numkeys` arg is 0.
-     - issue: https://github.com/redis/redis/issues/9493
-     - fix: https://github.com/redis/redis/pull/9733
-
-    So, don't use this with EVAL or EVALSHA.
-    """
-
-    __slots__ = ("commands", "node")
-
-    def __init__(self) -> None:
-        self.commands: Dict[str, Union[int, Dict[str, Any]]] = {}
-
-    async def initialize(self, node: Optional["ClusterNode"] = None) -> None:
-        if node:
-            self.node = node
-
-        commands = await self.node.execute_command("COMMAND")
-        for cmd, command in commands.items():
-            if "movablekeys" in command["flags"]:
-                commands[cmd] = -1
-            elif command["first_key_pos"] == 0 and command["last_key_pos"] == 0:
-                commands[cmd] = 0
-            elif command["first_key_pos"] == 1 and command["last_key_pos"] == 1:
-                commands[cmd] = 1
-        self.commands = {cmd.upper(): command for cmd, command in commands.items()}
-
-    # As soon as this PR is merged into Redis, we should reimplement
-    # our logic to use COMMAND INFO changes to determine the key positions
-    # https://github.com/redis/redis/pull/8324
-    async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
-        if len(args) < 2:
-            # The command has no keys in it
-            return None
-
-        try:
-            command = self.commands[args[0]]
-        except KeyError:
-            # try to split the command name and to take only the main command
-            # e.g. 'memory' for 'memory usage'
-            args = args[0].split() + list(args[1:])
-            cmd_name = args[0].upper()
-            if cmd_name not in self.commands:
-                # We'll try to reinitialize the commands cache, if the engine
-                # version has changed, the commands may not be current
-                await self.initialize()
-                if cmd_name not in self.commands:
-                    raise RedisError(
-                        f"{cmd_name} command doesn't exist in Redis commands"
-                    )
-
-            command = self.commands[cmd_name]
-
-        if command == 1:
-            return (args[1],)
-        if command == 0:
-            return None
-        if command == -1:
-            return await self._get_moveable_keys(*args)
-
-        last_key_pos = command["last_key_pos"]
-        if last_key_pos < 0:
-            last_key_pos = len(args) + last_key_pos
-        return args[command["first_key_pos"] : last_key_pos + 1 : command["step_count"]]
-
-    async def _get_moveable_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
-        try:
-            keys = await self.node.execute_command("COMMAND GETKEYS", *args)
-        except ResponseError as e:
-            message = e.__str__()
-            if (
-                "Invalid arguments" in message
-                or "The command has no key arguments" in message
-            ):
-                return None
-            else:
-                raise e
-        return keys
diff --git a/redis/cluster.py b/redis/cluster.py
index 91deaea..245dfdb 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -8,7 +8,7 @@ from typing import Any, Callable, Dict, List, Optional, Tuple, Union
 
 from redis.backoff import default_backoff
 from redis.client import CaseInsensitiveDict, PubSub, Redis, parse_scan
-from redis.commands import READ_COMMANDS, CommandsParser, RedisClusterCommands
+from redis.commands import READ_COMMANDS, RedisClusterCommands
 from redis.connection import ConnectionPool, DefaultParser, Encoder, parse_url
 from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot
 from redis.exceptions import (
@@ -29,6 +29,7 @@ from redis.exceptions import (
     TryAgainError,
 )
 from redis.lock import Lock
+from redis.parsers import CommandsParser, Encoder
 from redis.retry import Retry
 from redis.utils import (
     dict_merge,
diff --git a/redis/commands/__init__.py b/redis/commands/__init__.py
index f3f0828..a94d976 100644
--- a/redis/commands/__init__.py
+++ b/redis/commands/__init__.py
@@ -1,7 +1,6 @@
 from .cluster import READ_COMMANDS, AsyncRedisClusterCommands, RedisClusterCommands
 from .core import AsyncCoreCommands, CoreCommands
 from .helpers import list_or_args
-from .parser import CommandsParser
 from .redismodules import AsyncRedisModuleCommands, RedisModuleCommands
 from .sentinel import AsyncSentinelCommands, SentinelCommands
 
@@ -10,7 +9,6 @@ __all__ = [
     "AsyncRedisClusterCommands",
     "AsyncRedisModuleCommands",
     "AsyncSentinelCommands",
-    "CommandsParser",
     "CoreCommands",
     "READ_COMMANDS",
     "RedisClusterCommands",
diff --git a/redis/connection.py b/redis/connection.py
index 9c5b536..568c11c 100755
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -1,66 +1,41 @@
 import copy
-import errno
-import io
 import os
 import socket
+import ssl
 import threading
 import weakref
 from itertools import chain
 from queue import Empty, Full, LifoQueue
 from time import time
-from typing import Optional
+from typing import Optional, Type, Union
 from urllib.parse import parse_qs, unquote, urlparse
 
-from redis.backoff import NoBackoff
-from redis.credentials import CredentialProvider, UsernamePasswordCredentialProvider
-from redis.exceptions import (
+from .backoff import NoBackoff
+from .credentials import CredentialProvider, UsernamePasswordCredentialProvider
+from .exceptions import (
     AuthenticationError,
     AuthenticationWrongNumberOfArgsError,
-    BusyLoadingError,
     ChildDeadlockedError,
     ConnectionError,
     DataError,
-    ExecAbortError,
-    InvalidResponse,
-    ModuleError,
-    NoPermissionError,
-    NoScriptError,
-    ReadOnlyError,
     RedisError,
     ResponseError,
     TimeoutError,
 )
-from redis.retry import Retry
-from redis.utils import CRYPTOGRAPHY_AVAILABLE, HIREDIS_AVAILABLE, str_if_bytes
-
-try:
-    import ssl
-
-    ssl_available = True
-except ImportError:
-    ssl_available = False
-
-NONBLOCKING_EXCEPTION_ERROR_NUMBERS = {BlockingIOError: errno.EWOULDBLOCK}
-
-if ssl_available:
-    if hasattr(ssl, "SSLWantReadError"):
-        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantReadError] = 2
-        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantWriteError] = 2
-    else:
-        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLError] = 2
-
-NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys())
-
-if HIREDIS_AVAILABLE:
-    import hiredis
+from .parsers import Encoder, HiredisParser, RESP2Parser
+from .retry import Retry
+from .utils import (
+    CRYPTOGRAPHY_AVAILABLE,
+    HIREDIS_AVAILABLE,
+    SSL_AVAILABLE,
+    str_if_bytes,
+)
 
 SYM_STAR = b"*"
 SYM_DOLLAR = b"$"
 SYM_CRLF = b"\r\n"
 SYM_EMPTY = b""
 
-SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server."
-
 SENTINEL = object()
 MODULE_LOAD_ERROR = "Error loading the extension. " "Please check the server logs."
 NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name"
@@ -81,401 +56,11 @@ NO_AUTH_SET_ERROR = {
 }
 
 
-class Encoder:
-    "Encode strings to bytes-like and decode bytes-like to strings"
-
-    def __init__(self, encoding, encoding_errors, decode_responses):
-        self.encoding = encoding
-        self.encoding_errors = encoding_errors
-        self.decode_responses = decode_responses
-
-    def encode(self, value):
-        "Return a bytestring or bytes-like representation of the value"
-        if isinstance(value, (bytes, memoryview)):
-            return value
-        elif isinstance(value, bool):
-            # special case bool since it is a subclass of int
-            raise DataError(
-                "Invalid input of type: 'bool'. Convert to a "
-                "bytes, string, int or float first."
-            )
-        elif isinstance(value, (int, float)):
-            value = repr(value).encode()
-        elif not isinstance(value, str):
-            # a value we don't know how to deal with. throw an error
-            typename = type(value).__name__
-            raise DataError(
-                f"Invalid input of type: '{typename}'. "
-                f"Convert to a bytes, string, int or float first."
-            )
-        if isinstance(value, str):
-            value = value.encode(self.encoding, self.encoding_errors)
-        return value
-
-    def decode(self, value, force=False):
-        "Return a unicode string from the bytes-like representation"
-        if self.decode_responses or force:
-            if isinstance(value, memoryview):
-                value = value.tobytes()
-            if isinstance(value, bytes):
-                value = value.decode(self.encoding, self.encoding_errors)
-        return value
-
-
-class BaseParser:
-    EXCEPTION_CLASSES = {
-        "ERR": {
-            "max number of clients reached": ConnectionError,
-            "invalid password": AuthenticationError,
-            # some Redis server versions report invalid command syntax
-            # in lowercase
-            "wrong number of arguments "
-            "for 'auth' command": AuthenticationWrongNumberOfArgsError,
-            # some Redis server versions report invalid command syntax
-            # in uppercase
-            "wrong number of arguments "
-            "for 'AUTH' command": AuthenticationWrongNumberOfArgsError,
-            MODULE_LOAD_ERROR: ModuleError,
-            MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError,
-            NO_SUCH_MODULE_ERROR: ModuleError,
-            MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,
-            **NO_AUTH_SET_ERROR,
-        },
-        "WRONGPASS": AuthenticationError,
-        "EXECABORT": ExecAbortError,
-        "LOADING": BusyLoadingError,
-        "NOSCRIPT": NoScriptError,
-        "READONLY": ReadOnlyError,
-        "NOAUTH": AuthenticationError,
-        "NOPERM": NoPermissionError,
-    }
-
-    def parse_error(self, response):
-        "Parse an error response"
-        error_code = response.split(" ")[0]
-        if error_code in self.EXCEPTION_CLASSES:
-            response = response[len(error_code) + 1 :]
-            exception_class = self.EXCEPTION_CLASSES[error_code]
-            if isinstance(exception_class, dict):
-                exception_class = exception_class.get(response, ResponseError)
-            return exception_class(response)
-        return ResponseError(response)
-
-
-class SocketBuffer:
-    def __init__(self, socket, socket_read_size, socket_timeout):
-        self._sock = socket
-        self.socket_read_size = socket_read_size
-        self.socket_timeout = socket_timeout
-        self._buffer = io.BytesIO()
-        # number of bytes written to the buffer from the socket
-        self.bytes_written = 0
-        # number of bytes read from the buffer
-        self.bytes_read = 0
-
-    @property
-    def length(self):
-        return self.bytes_written - self.bytes_read
-
-    def _read_from_socket(self, length=None, timeout=SENTINEL, raise_on_timeout=True):
-        sock = self._sock
-        socket_read_size = self.socket_read_size
-        buf = self._buffer
-        buf.seek(self.bytes_written)
-        marker = 0
-        custom_timeout = timeout is not SENTINEL
-
-        try:
-            if custom_timeout:
-                sock.settimeout(timeout)
-            while True:
-                data = self._sock.recv(socket_read_size)
-                # an empty string indicates the server shutdown the socket
-                if isinstance(data, bytes) and len(data) == 0:
-                    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-                buf.write(data)
-                data_length = len(data)
-                self.bytes_written += data_length
-                marker += data_length
-
-                if length is not None and length > marker:
-                    continue
-                return True
-        except socket.timeout:
-            if raise_on_timeout:
-                raise TimeoutError("Timeout reading from socket")
-            return False
-        except NONBLOCKING_EXCEPTIONS as ex:
-            # if we're in nonblocking mode and the recv raises a
-            # blocking error, simply return False indicating that
-            # there's no data to be read. otherwise raise the
-            # original exception.
-            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
-            if not raise_on_timeout and ex.errno == allowed:
-                return False
-            raise ConnectionError(f"Error while reading from socket: {ex.args}")
-        finally:
-            if custom_timeout:
-                sock.settimeout(self.socket_timeout)
-
-    def can_read(self, timeout):
-        return bool(self.length) or self._read_from_socket(
-            timeout=timeout, raise_on_timeout=False
-        )
-
-    def read(self, length):
-        length = length + 2  # make sure to read the \r\n terminator
-        # make sure we've read enough data from the socket
-        if length > self.length:
-            self._read_from_socket(length - self.length)
-
-        self._buffer.seek(self.bytes_read)
-        data = self._buffer.read(length)
-        self.bytes_read += len(data)
-
-        # purge the buffer when we've consumed it all so it doesn't
-        # grow forever
-        if self.bytes_read == self.bytes_written:
-            self.purge()
-
-        return data[:-2]
-
-    def readline(self):
-        buf = self._buffer
-        buf.seek(self.bytes_read)
-        data = buf.readline()
-        while not data.endswith(SYM_CRLF):
-            # there's more data in the socket that we need
-            self._read_from_socket()
-            buf.seek(self.bytes_read)
-            data = buf.readline()
-
-        self.bytes_read += len(data)
-
-        # purge the buffer when we've consumed it all so it doesn't
-        # grow forever
-        if self.bytes_read == self.bytes_written:
-            self.purge()
-
-        return data[:-2]
-
-    def purge(self):
-        self._buffer.seek(0)
-        self._buffer.truncate()
-        self.bytes_written = 0
-        self.bytes_read = 0
-
-    def close(self):
-        try:
-            self.purge()
-            self._buffer.close()
-        except Exception:
-            # issue #633 suggests the purge/close somehow raised a
-            # BadFileDescriptor error. Perhaps the client ran out of
-            # memory or something else? It's probably OK to ignore
-            # any error being raised from purge/close since we're
-            # removing the reference to the instance below.
-            pass
-        self._buffer = None
-        self._sock = None
-
-
-class PythonParser(BaseParser):
-    "Plain Python parsing class"
-
-    def __init__(self, socket_read_size):
-        self.socket_read_size = socket_read_size
-        self.encoder = None
-        self._sock = None
-        self._buffer = None
-
-    def __del__(self):
-        try:
-            self.on_disconnect()
-        except Exception:
-            pass
-
-    def on_connect(self, connection):
-        "Called when the socket connects"
-        self._sock = connection._sock
-        self._buffer = SocketBuffer(
-            self._sock, self.socket_read_size, connection.socket_timeout
-        )
-        self.encoder = connection.encoder
-
-    def on_disconnect(self):
-        "Called when the socket disconnects"
-        self._sock = None
-        if self._buffer is not None:
-            self._buffer.close()
-            self._buffer = None
-        self.encoder = None
-
-    def can_read(self, timeout):
-        return self._buffer and self._buffer.can_read(timeout)
-
-    def read_response(self, disable_decoding=False):
-        raw = self._buffer.readline()
-        if not raw:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-
-        byte, response = raw[:1], raw[1:]
-
-        if byte not in (b"-", b"+", b":", b"$", b"*"):
-            raise InvalidResponse(f"Protocol Error: {raw!r}")
-
-        # server returned an error
-        if byte == b"-":
-            response = response.decode("utf-8", errors="replace")
-            error = self.parse_error(response)
-            # if the error is a ConnectionError, raise immediately so the user
-            # is notified
-            if isinstance(error, ConnectionError):
-                raise error
-            # otherwise, we're dealing with a ResponseError that might belong
-            # inside a pipeline response. the connection's read_response()
-            # and/or the pipeline's execute() will raise this error if
-            # necessary, so just return the exception instance here.
-            return error
-        # single value
-        elif byte == b"+":
-            pass
-        # int value
-        elif byte == b":":
-            response = int(response)
-        # bulk response
-        elif byte == b"$":
-            length = int(response)
-            if length == -1:
-                return None
-            response = self._buffer.read(length)
-        # multi-bulk response
-        elif byte == b"*":
-            length = int(response)
-            if length == -1:
-                return None
-            response = [
-                self.read_response(disable_decoding=disable_decoding)
-                for i in range(length)
-            ]
-        if isinstance(response, bytes) and disable_decoding is False:
-            response = self.encoder.decode(response)
-        return response
-
-
-class HiredisParser(BaseParser):
-    "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)
-
-    def __del__(self):
-        try:
-            self.on_disconnect()
-        except Exception:
-            pass
-
-    def on_connect(self, connection, **kwargs):
-        self._sock = connection._sock
-        self._socket_timeout = connection.socket_timeout
-        kwargs = {
-            "protocolError": InvalidResponse,
-            "replyError": self.parse_error,
-            "errors": connection.encoder.encoding_errors,
-        }
-
-        if connection.encoder.decode_responses:
-            kwargs["encoding"] = connection.encoder.encoding
-        self._reader = hiredis.Reader(**kwargs)
-        self._next_response = False
-
-    def on_disconnect(self):
-        self._sock = None
-        self._reader = None
-        self._next_response = False
-
-    def can_read(self, timeout):
-        if not self._reader:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-
-        if self._next_response is False:
-            self._next_response = self._reader.gets()
-            if self._next_response is False:
-                return self.read_from_socket(timeout=timeout, raise_on_timeout=False)
-        return True
-
-    def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True):
-        sock = self._sock
-        custom_timeout = timeout is not SENTINEL
-        try:
-            if custom_timeout:
-                sock.settimeout(timeout)
-            bufflen = self._sock.recv_into(self._buffer)
-            if bufflen == 0:
-                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-            self._reader.feed(self._buffer, 0, bufflen)
-            # data was read from the socket and added to the buffer.
-            # return True to indicate that data was read.
-            return True
-        except socket.timeout:
-            if raise_on_timeout:
-                raise TimeoutError("Timeout reading from socket")
-            return False
-        except NONBLOCKING_EXCEPTIONS as ex:
-            # if we're in nonblocking mode and the recv raises a
-            # blocking error, simply return False indicating that
-            # there's no data to be read. otherwise raise the
-            # original exception.
-            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
-            if not raise_on_timeout and ex.errno == allowed:
-                return False
-            raise ConnectionError(f"Error while reading from socket: {ex.args}")
-        finally:
-            if custom_timeout:
-                sock.settimeout(self._socket_timeout)
-
-    def read_response(self, disable_decoding=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 False:
-            response = self._next_response
-            self._next_response = False
-            return response
-
-        if disable_decoding:
-            response = self._reader.gets(False)
-        else:
-            response = self._reader.gets()
-
-        while response is False:
-            self.read_from_socket()
-            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
-        # happened
-        if isinstance(response, ConnectionError):
-            raise response
-        elif (
-            isinstance(response, list)
-            and response
-            and isinstance(response[0], ConnectionError)
-        ):
-            raise response[0]
-        return response
-
-
+DefaultParser: Type[Union[RESP2Parser, HiredisParser]]
 if HIREDIS_AVAILABLE:
     DefaultParser = HiredisParser
 else:
-    DefaultParser = PythonParser
+    DefaultParser = RESP2Parser
 
 
 class Connection:
@@ -944,7 +529,7 @@ class SSLConnection(Connection):
         Raises:
             RedisError
         """  # noqa
-        if not ssl_available:
+        if not SSL_AVAILABLE:
             raise RedisError("Python wasn't built with SSL support")
 
         super().__init__(**kwargs)
diff --git a/redis/parsers/__init__.py b/redis/parsers/__init__.py
new file mode 100644
index 0000000..c6d950d
--- /dev/null
+++ b/redis/parsers/__init__.py
@@ -0,0 +1,16 @@
+from .commands import AsyncCommandsParser, CommandsParser
+from .encoders import Encoder
+from .hiredis import AsyncHiredisParser, HiredisParser
+from .protocol import BaseParser
+from .resp2 import AsyncRESP2Parser, RESP2Parser
+
+__all__ = [
+    "AsyncCommandsParser",
+    "AsyncHiredisParser",
+    "AsyncRESP2Parser",
+    "CommandsParser",
+    "Encoder",
+    "BaseParser",
+    "HiredisParser",
+    "RESP2Parser",
+]
diff --git a/redis/commands/parser.py b/redis/parsers/commands.py
index 115230a..7981a52 100644
--- a/redis/commands/parser.py
+++ b/redis/parsers/commands.py
@@ -1,6 +1,99 @@
+from typing import TYPE_CHECKING, Any, Dict, Optional, Tuple, Union
+
 from redis.exceptions import RedisError, ResponseError
 from redis.utils import str_if_bytes
 
+if TYPE_CHECKING:
+    from redis.asyncio.cluster import ClusterNode
+
+
+class AsyncCommandsParser:
+    """
+    Parses Redis commands to get command keys.
+
+    COMMAND output is used to determine key locations.
+    Commands that do not have a predefined key location are flagged with 'movablekeys',
+    and these commands' keys are determined by the command 'COMMAND GETKEYS'.
+
+    NOTE: Due to a bug in redis<7.0, this does not work properly
+    for EVAL or EVALSHA when the `numkeys` arg is 0.
+     - issue: https://github.com/redis/redis/issues/9493
+     - fix: https://github.com/redis/redis/pull/9733
+
+    So, don't use this with EVAL or EVALSHA.
+    """
+
+    __slots__ = ("commands", "node")
+
+    def __init__(self) -> None:
+        self.commands: Dict[str, Union[int, Dict[str, Any]]] = {}
+
+    async def initialize(self, node: Optional["ClusterNode"] = None) -> None:
+        if node:
+            self.node = node
+
+        commands = await self.node.execute_command("COMMAND")
+        for cmd, command in commands.items():
+            if "movablekeys" in command["flags"]:
+                commands[cmd] = -1
+            elif command["first_key_pos"] == 0 and command["last_key_pos"] == 0:
+                commands[cmd] = 0
+            elif command["first_key_pos"] == 1 and command["last_key_pos"] == 1:
+                commands[cmd] = 1
+        self.commands = {cmd.upper(): command for cmd, command in commands.items()}
+
+    # As soon as this PR is merged into Redis, we should reimplement
+    # our logic to use COMMAND INFO changes to determine the key positions
+    # https://github.com/redis/redis/pull/8324
+    async def get_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
+        if len(args) < 2:
+            # The command has no keys in it
+            return None
+
+        try:
+            command = self.commands[args[0]]
+        except KeyError:
+            # try to split the command name and to take only the main command
+            # e.g. 'memory' for 'memory usage'
+            args = args[0].split() + list(args[1:])
+            cmd_name = args[0].upper()
+            if cmd_name not in self.commands:
+                # We'll try to reinitialize the commands cache, if the engine
+                # version has changed, the commands may not be current
+                await self.initialize()
+                if cmd_name not in self.commands:
+                    raise RedisError(
+                        f"{cmd_name} command doesn't exist in Redis commands"
+                    )
+
+            command = self.commands[cmd_name]
+
+        if command == 1:
+            return (args[1],)
+        if command == 0:
+            return None
+        if command == -1:
+            return await self._get_moveable_keys(*args)
+
+        last_key_pos = command["last_key_pos"]
+        if last_key_pos < 0:
+            last_key_pos = len(args) + last_key_pos
+        return args[command["first_key_pos"] : last_key_pos + 1 : command["step_count"]]
+
+    async def _get_moveable_keys(self, *args: Any) -> Optional[Tuple[str, ...]]:
+        try:
+            keys = await self.node.execute_command("COMMAND GETKEYS", *args)
+        except ResponseError as e:
+            message = e.__str__()
+            if (
+                "Invalid arguments" in message
+                or "The command has no key arguments" in message
+            ):
+                return None
+            else:
+                raise e
+        return keys
+
 
 class CommandsParser:
     """
diff --git a/redis/parsers/encoders.py b/redis/parsers/encoders.py
new file mode 100644
index 0000000..85fb292
--- /dev/null
+++ b/redis/parsers/encoders.py
@@ -0,0 +1,44 @@
+from redis.typing import EncodableT, EncodedT
+
+from ..exceptions import DataError
+
+
+class Encoder:
+    "Encode strings to bytes-like and decode bytes-like to strings"
+
+    def __init__(self, encoding, encoding_errors, decode_responses):
+        self.encoding = encoding
+        self.encoding_errors = encoding_errors
+        self.decode_responses = decode_responses
+
+    def encode(self, value):
+        "Return a bytestring or bytes-like representation of the value"
+        if isinstance(value, (bytes, memoryview)):
+            return value
+        elif isinstance(value, (int, float)):
+            value = repr(value).encode()
+        elif isinstance(value, str):
+            value = value.encode(self.encoding, self.encoding_errors)
+        elif isinstance(value, bool):
+            # special case bool since it is a subclass of int
+            raise DataError(
+                "Invalid input of type: 'bool'. Convert to a "
+                "bytes, string, int or float first."
+            )
+        elif not isinstance(value, str):
+            # a value we don't know how to deal with. throw an error
+            typename = type(value).__name__
+            raise DataError(
+                f"Invalid input of type: '{typename}'. "
+                f"Convert to a bytes, string, int or float first."
+            )
+        return value
+
+    def decode(self, value, force=False):
+        "Return a unicode string from the bytes-like representation"
+        if self.decode_responses or force:
+            if isinstance(value, memoryview):
+                value = value.tobytes()
+            if isinstance(value, bytes):
+                value = value.decode(self.encoding, self.encoding_errors)
+        return value
diff --git a/redis/parsers/hiredis.py b/redis/parsers/hiredis.py
new file mode 100644
index 0000000..44d5b81
--- /dev/null
+++ b/redis/parsers/hiredis.py
@@ -0,0 +1,217 @@
+import asyncio
+import socket
+from typing import Callable, List, Optional, TypedDict, Union
+
+import async_timeout
+
+from ..exceptions import (
+    AuthenticationError,
+    ConnectionError,
+    InvalidResponse,
+    RedisError,
+)
+from ..typing import EncodableT
+from ..utils import HIREDIS_AVAILABLE
+from .protocol import AsyncBaseParser, BaseParser
+from .socket import (
+    NONBLOCKING_EXCEPTION_ERROR_NUMBERS,
+    NONBLOCKING_EXCEPTIONS,
+    SENTINEL,
+    SERVER_CLOSED_CONNECTION_ERROR,
+)
+
+
+class _HiredisReaderArgs(TypedDict, total=False):
+    protocolError: Callable[[str], Exception]
+    replyError: Callable[[str], Exception]
+    encoding: Optional[str]
+    errors: Optional[str]
+
+
+class HiredisParser(BaseParser):
+    "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)
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_connect(self, connection, **kwargs):
+        import hiredis
+
+        self._sock = connection._sock
+        self._socket_timeout = connection.socket_timeout
+        kwargs = {
+            "protocolError": InvalidResponse,
+            "replyError": self.parse_error,
+            "errors": connection.encoder.encoding_errors,
+        }
+
+        if connection.encoder.decode_responses:
+            kwargs["encoding"] = connection.encoder.encoding
+        self._reader = hiredis.Reader(**kwargs)
+        self._next_response = False
+
+    def on_disconnect(self):
+        self._sock = None
+        self._reader = None
+        self._next_response = False
+
+    def can_read(self, timeout):
+        if not self._reader:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+
+        if self._next_response is False:
+            self._next_response = self._reader.gets()
+            if self._next_response is False:
+                return self.read_from_socket(timeout=timeout, raise_on_timeout=False)
+        return True
+
+    def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True):
+        sock = self._sock
+        custom_timeout = timeout is not SENTINEL
+        try:
+            if custom_timeout:
+                sock.settimeout(timeout)
+            bufflen = self._sock.recv_into(self._buffer)
+            if bufflen == 0:
+                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+            self._reader.feed(self._buffer, 0, bufflen)
+            # data was read from the socket and added to the buffer.
+            # return True to indicate that data was read.
+            return True
+        except socket.timeout:
+            if raise_on_timeout:
+                raise TimeoutError("Timeout reading from socket")
+            return False
+        except NONBLOCKING_EXCEPTIONS as ex:
+            # if we're in nonblocking mode and the recv raises a
+            # blocking error, simply return False indicating that
+            # there's no data to be read. otherwise raise the
+            # original exception.
+            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
+            if not raise_on_timeout and ex.errno == allowed:
+                return False
+            raise ConnectionError(f"Error while reading from socket: {ex.args}")
+        finally:
+            if custom_timeout:
+                sock.settimeout(self._socket_timeout)
+
+    def read_response(self, disable_decoding=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 False:
+            response = self._next_response
+            self._next_response = False
+            return response
+
+        if disable_decoding:
+            response = self._reader.gets(False)
+        else:
+            response = self._reader.gets()
+
+        while response is False:
+            self.read_from_socket()
+            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
+        # happened
+        if isinstance(response, (AuthenticationError, ConnectionError)):
+            raise response
+        elif (
+            isinstance(response, list)
+            and response
+            and isinstance(response[0], (AuthenticationError, ConnectionError))
+        ):
+            raise response[0]
+        return response
+
+
+class AsyncHiredisParser(AsyncBaseParser):
+    """PAsync implementation of parser class for connections using Hiredis"""
+
+    __slots__ = AsyncBaseParser.__slots__ + ("_reader",)
+
+    def __init__(self, socket_read_size: int):
+        if not HIREDIS_AVAILABLE:
+            raise RedisError("Hiredis is not available.")
+
+        import hiredis
+
+        super().__init__(socket_read_size=socket_read_size)
+        self._reader: Optional[hiredis.Reader] = None
+
+    def on_connect(self, connection: "Connection"):
+        import hiredis
+
+        self._stream = connection._reader
+        kwargs: _HiredisReaderArgs = {
+            "protocolError": InvalidResponse,
+            "replyError": self.parse_error,
+        }
+        if connection.encoder.decode_responses:
+            kwargs["encoding"] = connection.encoder.encoding
+            kwargs["errors"] = connection.encoder.encoding_errors
+
+        self._reader = hiredis.Reader(**kwargs)
+
+    def on_disconnect(self):
+        self._stream = None
+        self._reader = None
+
+    async def can_read_destructive(self):
+        if not self._stream or not self._reader:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        if self._reader.gets():
+            return True
+        try:
+            async with async_timeout.timeout(0):
+                return await self.read_from_socket()
+        except asyncio.TimeoutError:
+            return False
+
+    async def read_from_socket(self):
+        buffer = await self._stream.read(self._read_size)
+        if not buffer or not isinstance(buffer, bytes):
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
+        self._reader.feed(buffer)
+        # data was read from the socket and added to the buffer.
+        # return True to indicate that data was read.
+        return True
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, List[EncodableT]]:
+        if not self._stream or not self._reader:
+            self.on_disconnect()
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from None
+
+        response = self._reader.gets()
+        while response is False:
+            await self.read_from_socket()
+            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
+        # happened
+        if isinstance(response, (AuthenticationError, ConnectionError)):
+            raise response
+        elif (
+            isinstance(response, list)
+            and response
+            and isinstance(response[0], (AuthenticationError, ConnectionError))
+        ):
+            raise response[0]
+        return response
diff --git a/redis/parsers/protocol.py b/redis/parsers/protocol.py
new file mode 100644
index 0000000..6faaddd
--- /dev/null
+++ b/redis/parsers/protocol.py
@@ -0,0 +1,107 @@
+from abc import ABC
+from asyncio import StreamReader
+from typing import List, Optional, Union
+
+from redis.typing import EncodableT
+
+from ..exceptions import (
+    AuthenticationError,
+    AuthenticationWrongNumberOfArgsError,
+    BusyLoadingError,
+    ConnectionError,
+    ExecAbortError,
+    ModuleError,
+    NoPermissionError,
+    NoScriptError,
+    ReadOnlyError,
+    ResponseError,
+)
+
+MODULE_LOAD_ERROR = "Error loading the extension. " "Please check the server logs."
+NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name"
+MODULE_UNLOAD_NOT_POSSIBLE_ERROR = "Error unloading module: operation not " "possible."
+MODULE_EXPORTS_DATA_TYPES_ERROR = (
+    "Error unloading module: the module "
+    "exports one or more module-side data "
+    "types, can't unload"
+)
+# user send an AUTH cmd to a server without authorization configured
+NO_AUTH_SET_ERROR = {
+    # Redis >= 6.0
+    "AUTH <password> called without any password "
+    "configured for the default user. Are you sure "
+    "your configuration is correct?": AuthenticationError,
+    # Redis < 6.0
+    "Client sent AUTH, but no password is set": AuthenticationError,
+}
+
+
+class BaseParser(ABC):
+
+    EXCEPTION_CLASSES = {
+        "ERR": {
+            "max number of clients reached": ConnectionError,
+            "invalid password": AuthenticationError,
+            # some Redis server versions report invalid command syntax
+            # in lowercase
+            "wrong number of arguments "
+            "for 'auth' command": AuthenticationWrongNumberOfArgsError,
+            # some Redis server versions report invalid command syntax
+            # in uppercase
+            "wrong number of arguments "
+            "for 'AUTH' command": AuthenticationWrongNumberOfArgsError,
+            MODULE_LOAD_ERROR: ModuleError,
+            MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError,
+            NO_SUCH_MODULE_ERROR: ModuleError,
+            MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError,
+            **NO_AUTH_SET_ERROR,
+        },
+        "WRONGPASS": AuthenticationError,
+        "EXECABORT": ExecAbortError,
+        "LOADING": BusyLoadingError,
+        "NOSCRIPT": NoScriptError,
+        "READONLY": ReadOnlyError,
+        "NOAUTH": AuthenticationError,
+        "NOPERM": NoPermissionError,
+    }
+
+    def parse_error(self, response):
+        "Parse an error response"
+        error_code = response.split(" ")[0]
+        if error_code in self.EXCEPTION_CLASSES:
+            response = response[len(error_code) + 1 :]
+            exception_class = self.EXCEPTION_CLASSES[error_code]
+            if isinstance(exception_class, dict):
+                exception_class = exception_class.get(response, ResponseError)
+            return exception_class(response)
+        return ResponseError(response)
+
+
+class AsyncBaseParser(BaseParser):
+    """Base parsing class for the python-backed async parser"""
+
+    __slots__ = "_stream", "_read_size"
+
+    def __init__(self, socket_read_size: int):
+        self._stream: Optional[StreamReader] = None
+        self._read_size = socket_read_size
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_disconnect(self):
+        raise NotImplementedError()
+
+    def on_connect(self, connection: "Connection"):
+        raise NotImplementedError()
+
+    async def can_read_destructive(self) -> bool:
+        raise NotImplementedError()
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, ResponseError, None, List[EncodableT]]:
+        raise NotImplementedError()
diff --git a/redis/parsers/resp2.py b/redis/parsers/resp2.py
new file mode 100644
index 0000000..32872ea
--- /dev/null
+++ b/redis/parsers/resp2.py
@@ -0,0 +1,203 @@
+from asyncio import IncompleteReadError, TimeoutError
+from typing import Any, Optional, Union
+
+import async_timeout
+
+from redis.typing import EncodableT
+
+from ..exceptions import ConnectionError, InvalidResponse, RedisError, ResponseError
+from .encoders import Encoder
+from .protocol import AsyncBaseParser, BaseParser
+from .socket import SERVER_CLOSED_CONNECTION_ERROR, SocketBuffer
+
+
+class RESP2Parser(BaseParser):
+    "Plain Python parsing class"
+
+    def __init__(self, socket_read_size):
+        self.socket_read_size = socket_read_size
+        self.encoder = None
+        self._sock = None
+        self._buffer = None
+
+    def __del__(self):
+        try:
+            self.on_disconnect()
+        except Exception:
+            pass
+
+    def on_connect(self, connection):
+        "Called when the socket connects"
+        self._sock = connection._sock
+        self._buffer = SocketBuffer(
+            self._sock, self.socket_read_size, connection.socket_timeout
+        )
+        self.encoder = connection.encoder
+
+    def on_disconnect(self):
+        "Called when the socket disconnects"
+        self._sock = None
+        if self._buffer is not None:
+            self._buffer.close()
+            self._buffer = None
+        self.encoder = None
+
+    def can_read(self, timeout):
+        return self._buffer and self._buffer.can_read(timeout)
+
+    def read_response(self, disable_decoding=False):
+        raw = self._buffer.readline()
+        if not raw:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+
+        byte, response = raw[:1], raw[1:]
+
+        if byte not in (b"-", b"+", b":", b"$", b"*"):
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        # server returned an error
+        if byte == b"-":
+            response = response.decode("utf-8", errors="replace")
+            error = self.parse_error(response)
+            # if the error is a ConnectionError, raise immediately so the user
+            # is notified
+            if isinstance(error, ConnectionError):
+                raise error
+            # otherwise, we're dealing with a ResponseError that might belong
+            # inside a pipeline response. the connection's read_response()
+            # and/or the pipeline's execute() will raise this error if
+            # necessary, so just return the exception instance here.
+            return error
+        # single value
+        elif byte == b"+":
+            pass
+        # int value
+        elif byte == b":":
+            response = int(response)
+        # bulk response
+        elif byte == b"$":
+            length = int(response)
+            if length == -1:
+                return None
+            response = self._buffer.read(length)
+        # multi-bulk response
+        elif byte == b"*":
+            length = int(response)
+            if length == -1:
+                return None
+            response = [
+                self.read_response(disable_decoding=disable_decoding)
+                for i in range(length)
+            ]
+        if isinstance(response, bytes) and disable_decoding is False:
+            response = self.encoder.decode(response)
+        return response
+
+
+class AsyncRESP2Parser(AsyncBaseParser):
+    """Parsing class for the RESP2 protocol"""
+
+    __slots__ = AsyncBaseParser.__slots__ + ("encoder",)
+
+    def __init__(self, socket_read_size: int):
+        super().__init__(socket_read_size)
+        self.encoder: Optional[Encoder] = None
+
+    def on_connect(self, connection: "Connection"):
+        """Called when the stream connects"""
+        self._stream = connection._reader
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+
+        self.encoder = connection.encoder
+
+    def on_disconnect(self):
+        """Called when the stream disconnects"""
+        if self._stream is not None:
+            self._stream = None
+        self.encoder = None
+
+    async def can_read_destructive(self) -> bool:
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        try:
+            async with async_timeout.timeout(0):
+                return await self._stream.read(1)
+        except TimeoutError:
+            return False
+
+    async def read_response(
+        self, disable_decoding: bool = False
+    ) -> Union[EncodableT, ResponseError, None]:
+        if not self._stream or not self.encoder:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        raw = await self._readline()
+        if not raw:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        response: Any
+        byte, response = raw[:1], raw[1:]
+
+        if byte not in (b"-", b"+", b":", b"$", b"*"):
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        # server returned an error
+        if byte == b"-":
+            response = response.decode("utf-8", errors="replace")
+            error = self.parse_error(response)
+            # if the error is a ConnectionError, raise immediately so the user
+            # is notified
+            if isinstance(error, ConnectionError):
+                raise error
+            # otherwise, we're dealing with a ResponseError that might belong
+            # inside a pipeline response. the connection's read_response()
+            # and/or the pipeline's execute() will raise this error if
+            # necessary, so just return the exception instance here.
+            return error
+        # single value
+        elif byte == b"+":
+            pass
+        # int value
+        elif byte == b":":
+            response = int(response)
+        # bulk response
+        elif byte == b"$":
+            length = int(response)
+            if length == -1:
+                return None
+            response = await self._read(length)
+        # multi-bulk response
+        elif byte == b"*":
+            length = int(response)
+            if length == -1:
+                return None
+            response = [
+                (await self.read_response(disable_decoding)) for _ in range(length)
+            ]
+        if isinstance(response, bytes) and disable_decoding is False:
+            response = self.encoder.decode(response)
+        return response
+
+    async def _read(self, length: int) -> bytes:
+        """
+        Read `length` bytes of data.  These are assumed to be followed
+        by a '\r\n' terminator which is subsequently discarded.
+        """
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        try:
+            data = await self._stream.readexactly(length + 2)
+        except IncompleteReadError as error:
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error
+        return data[:-2]
+
+    async def _readline(self) -> bytes:
+        """
+        read an unknown number of bytes up to the next '\r\n'
+        line separator, which is discarded.
+        """
+        if self._stream is None:
+            raise RedisError("Buffer is closed.")
+        data = await self._stream.readline()
+        if not data.endswith(b"\r\n"):
+            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+        return data[:-2]
diff --git a/redis/parsers/socket.py b/redis/parsers/socket.py
new file mode 100644
index 0000000..b021051
--- /dev/null
+++ b/redis/parsers/socket.py
@@ -0,0 +1,140 @@
+import errno
+import io
+import socket
+
+from ..exceptions import ConnectionError, TimeoutError
+from ..utils import SSL_AVAILABLE
+
+NONBLOCKING_EXCEPTION_ERROR_NUMBERS = {BlockingIOError: errno.EWOULDBLOCK}
+
+if SSL_AVAILABLE:
+    import ssl
+
+    if hasattr(ssl, "SSLWantReadError"):
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantReadError] = 2
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantWriteError] = 2
+    else:
+        NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLError] = 2
+
+NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys())
+
+SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server."
+SENTINEL = object()
+
+
+class SocketBuffer:
+    def __init__(self, socket, socket_read_size, socket_timeout):
+        self._sock = socket
+        self.socket_read_size = socket_read_size
+        self.socket_timeout = socket_timeout
+        self._buffer = io.BytesIO()
+        # number of bytes written to the buffer from the socket
+        self.bytes_written = 0
+        # number of bytes read from the buffer
+        self.bytes_read = 0
+
+    @property
+    def length(self):
+        return self.bytes_written - self.bytes_read
+
+    def _read_from_socket(self, length=None, timeout=SENTINEL, raise_on_timeout=True):
+        sock = self._sock
+        socket_read_size = self.socket_read_size
+        buf = self._buffer
+        buf.seek(self.bytes_written)
+        marker = 0
+        custom_timeout = timeout is not SENTINEL
+
+        try:
+            if custom_timeout:
+                sock.settimeout(timeout)
+            while True:
+                data = self._sock.recv(socket_read_size)
+                # an empty string indicates the server shutdown the socket
+                if isinstance(data, bytes) and len(data) == 0:
+                    raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+                buf.write(data)
+                data_length = len(data)
+                self.bytes_written += data_length
+                marker += data_length
+
+                if length is not None and length > marker:
+                    continue
+                return True
+        except socket.timeout:
+            if raise_on_timeout:
+                raise TimeoutError("Timeout reading from socket")
+            return False
+        except NONBLOCKING_EXCEPTIONS as ex:
+            # if we're in nonblocking mode and the recv raises a
+            # blocking error, simply return False indicating that
+            # there's no data to be read. otherwise raise the
+            # original exception.
+            allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
+            if not raise_on_timeout and ex.errno == allowed:
+                return False
+            raise ConnectionError(f"Error while reading from socket: {ex.args}")
+        finally:
+            if custom_timeout:
+                sock.settimeout(self.socket_timeout)
+
+    def can_read(self, timeout):
+        return bool(self.length) or self._read_from_socket(
+            timeout=timeout, raise_on_timeout=False
+        )
+
+    def read(self, length):
+        length = length + 2  # make sure to read the \r\n terminator
+        # make sure we've read enough data from the socket
+        if length > self.length:
+            self._read_from_socket(length - self.length)
+
+        self._buffer.seek(self.bytes_read)
+        data = self._buffer.read(length)
+        self.bytes_read += len(data)
+
+        # purge the buffer when we've consumed it all so it doesn't
+        # grow forever
+        if self.bytes_read == self.bytes_written:
+            self.purge()
+
+        return data[:-2]
+
+    def readline(self):
+        buf = self._buffer
+        buf.seek(self.bytes_read)
+        data = buf.readline()
+        while not data.endswith(b"\r\n"):
+            # there's more data in the socket that we need
+            self._read_from_socket()
+            buf.seek(self.bytes_read)
+            data = buf.readline()
+
+        self.bytes_read += len(data)
+
+        # purge the buffer when we've consumed it all so it doesn't
+        # grow forever
+        if self.bytes_read == self.bytes_written:
+            self.purge()
+
+        return data[:-2]
+
+    def purge(self):
+        self._buffer.seek(0)
+        self._buffer.truncate()
+        self.bytes_written = 0
+        self.bytes_read = 0
+
+    def close(self):
+        try:
+            self.purge()
+            self._buffer.close()
+        except Exception:
+            # issue #633 suggests the purge/close somehow raised a
+            # BadFileDescriptor error. Perhaps the client ran out of
+            # memory or something else? It's probably OK to ignore
+            # any error being raised from purge/close since we're
+            # removing the reference to the instance below.
+            pass
+        self._buffer = None
+        self._sock = None
diff --git a/redis/typing.py b/redis/typing.py
index 8504c7d..c689604 100644
--- a/redis/typing.py
+++ b/redis/typing.py
@@ -1,14 +1,23 @@
 # from __future__ import annotations
 
 from datetime import datetime, timedelta
-from typing import TYPE_CHECKING, Any, Awaitable, Iterable, TypeVar, Union
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Awaitable,
+    Iterable,
+    Mapping,
+    Type,
+    TypeVar,
+    Union,
+)
 
 from redis.compat import Protocol
 
 if TYPE_CHECKING:
     from redis.asyncio.connection import ConnectionPool as AsyncConnectionPool
-    from redis.asyncio.connection import Encoder as AsyncEncoder
-    from redis.connection import ConnectionPool, Encoder
+    from redis.connection import ConnectionPool
+    from redis.parsers import Encoder
 
 
 Number = Union[int, float]
@@ -39,6 +48,8 @@ AnyKeyT = TypeVar("AnyKeyT", bytes, str, memoryview)
 AnyFieldT = TypeVar("AnyFieldT", bytes, str, memoryview)
 AnyChannelT = TypeVar("AnyChannelT", bytes, str, memoryview)
 
+ExceptionMappingT = Mapping[str, Union[Type[Exception], Mapping[str, Type[Exception]]]]
+
 
 class CommandsProtocol(Protocol):
     connection_pool: Union["AsyncConnectionPool", "ConnectionPool"]
@@ -48,7 +59,7 @@ class CommandsProtocol(Protocol):
 
 
 class ClusterCommandsProtocol(CommandsProtocol):
-    encoder: Union["AsyncEncoder", "Encoder"]
+    encoder: Union["Encoder", "Encoder"]
 
     def execute_command(self, *args, **options) -> Union[Any, Awaitable]:
         ...
diff --git a/redis/utils.py b/redis/utils.py
index 693d4e6..fa2d407 100644
--- a/redis/utils.py
+++ b/redis/utils.py
@@ -11,6 +11,13 @@ except ImportError:
     HIREDIS_AVAILABLE = False
 
 try:
+    import ssl
+
+    SSL_AVAILABLE = True
+except ImportError:
+    SSL_AVAILABLE = False
+
+try:
     import cryptography  # noqa
 
     CRYPTOGRAPHY_AVAILABLE = True
diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py
index 6982cc8..c011d60 100644
--- a/tests/test_asyncio/conftest.py
+++ b/tests/test_asyncio/conftest.py
@@ -9,14 +9,11 @@ from packaging.version import Version
 
 import redis.asyncio as redis
 from redis.asyncio.client import Monitor
-from redis.asyncio.connection import (
-    HIREDIS_AVAILABLE,
-    HiredisParser,
-    PythonParser,
-    parse_url,
-)
+from redis.asyncio.connection import parse_url
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
+from redis.parsers import AsyncHiredisParser, AsyncRESP2Parser
+from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import REDIS_INFO
 
 from .compat import mock
@@ -32,14 +29,14 @@ async def _get_info(redis_url):
 @pytest_asyncio.fixture(
     params=[
         pytest.param(
-            (True, PythonParser),
+            (True, AsyncRESP2Parser),
             marks=pytest.mark.skipif(
                 'config.REDIS_INFO["cluster_enabled"]', reason="cluster mode enabled"
             ),
         ),
-        (False, PythonParser),
+        (False, AsyncRESP2Parser),
         pytest.param(
-            (True, HiredisParser),
+            (True, AsyncHiredisParser),
             marks=[
                 pytest.mark.skipif(
                     'config.REDIS_INFO["cluster_enabled"]',
@@ -51,7 +48,7 @@ async def _get_info(redis_url):
             ],
         ),
         pytest.param(
-            (False, HiredisParser),
+            (False, AsyncHiredisParser),
             marks=pytest.mark.skipif(
                 not HIREDIS_AVAILABLE, reason="hiredis is not installed"
             ),
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 38bcaf6..5dee870 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -12,7 +12,6 @@ from _pytest.fixtures import FixtureRequest
 
 from redis.asyncio.cluster import ClusterNode, NodesManager, RedisCluster
 from redis.asyncio.connection import Connection, SSLConnection
-from redis.asyncio.parser import CommandsParser
 from redis.asyncio.retry import Retry
 from redis.backoff import ExponentialBackoff, NoBackoff, default_backoff
 from redis.cluster import PIPELINE_BLOCKED_COMMANDS, PRIMARY, REPLICA, get_node_name
@@ -29,6 +28,7 @@ from redis.exceptions import (
     RedisError,
     ResponseError,
 )
+from redis.parsers import AsyncCommandsParser
 from redis.utils import str_if_bytes
 from tests.conftest import (
     skip_if_redis_enterprise,
@@ -99,7 +99,7 @@ async def get_mocked_redis_client(*args, **kwargs) -> RedisCluster:
         execute_command_mock.side_effect = execute_command
 
         with mock.patch.object(
-            CommandsParser, "initialize", autospec=True
+            AsyncCommandsParser, "initialize", autospec=True
         ) as cmd_parser_initialize:
 
             def cmd_init_mock(self, r: ClusterNode) -> None:
@@ -549,7 +549,7 @@ class TestRedisClusterObj:
                     mocks["send_packed_command"].return_value = "MOCK_OK"
                     mocks["connect"].return_value = None
                     with mock.patch.object(
-                        CommandsParser, "initialize", autospec=True
+                        AsyncCommandsParser, "initialize", autospec=True
                     ) as cmd_parser_initialize:
 
                         def cmd_init_mock(self, r: ClusterNode) -> None:
@@ -2320,7 +2320,7 @@ class TestNodesManager:
             assert "Redis Cluster cannot be connected" in str(e.value)
 
             with mock.patch.object(
-                CommandsParser, "initialize", autospec=True
+                AsyncCommandsParser, "initialize", autospec=True
             ) as cmd_parser_initialize:
 
                 def cmd_init_mock(self, r: ClusterNode) -> None:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 6bf0034..6d0d54e 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -5,14 +5,11 @@ from unittest.mock import patch
 
 import pytest
 
-from redis.asyncio.connection import (
-    Connection,
-    PythonParser,
-    UnixDomainSocketConnection,
-)
+from redis.asyncio.connection import Connection, UnixDomainSocketConnection
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
 from redis.exceptions import ConnectionError, InvalidResponse, TimeoutError
+from redis.parsers import AsyncRESP2Parser
 from tests.conftest import skip_if_server_version_lt
 
 from .compat import mock
@@ -24,8 +21,8 @@ async def test_invalid_response(create_redis):
 
     raw = b"x"
 
-    parser: "PythonParser" = r.connection._parser
-    if not isinstance(parser, PythonParser):
+    parser: "AsyncRESP2Parser" = r.connection._parser
+    if not isinstance(parser, AsyncRESP2Parser):
         pytest.skip("PythonParser only")
     stream_mock = mock.Mock(parser._stream)
     stream_mock.readline.return_value = raw + b"\r\n"
diff --git a/tests/test_asyncio/test_encoding.py b/tests/test_asyncio/test_encoding.py
index 3efcf69..18cfdd8 100644
--- a/tests/test_asyncio/test_encoding.py
+++ b/tests/test_asyncio/test_encoding.py
@@ -95,9 +95,6 @@ class TestCommandsAreNotEncoded:
 
 
 class TestInvalidUserInput:
-    async def test_boolean_fails(self, r: redis.Redis):
-        with pytest.raises(DataError):
-            await r.set("a", True)  # type: ignore
 
     async def test_none_fails(self, r: redis.Redis):
         with pytest.raises(DataError):
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 6dedca9..0ca05e9 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -991,9 +991,9 @@ class TestBaseException:
         assert msg is not None
         # timeout waiting for another message which never arrives
         assert pubsub.connection.is_connected
-        with patch("redis.asyncio.connection.PythonParser.read_response") as mock1:
+        with patch("redis.parsers.AsyncRESP2Parser.read_response") as mock1:
             mock1.side_effect = BaseException("boom")
-            with patch("redis.asyncio.connection.HiredisParser.read_response") as mock2:
+            with patch("redis.parsers.AsyncHiredisParser.read_response") as mock2:
                 mock2.side_effect = BaseException("boom")
 
                 with pytest.raises(BaseException):
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index d18fbbb..4e920cf 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -17,7 +17,6 @@ from redis.cluster import (
     RedisCluster,
     get_node_name,
 )
-from redis.commands import CommandsParser
 from redis.connection import Connection
 from redis.crc import key_slot
 from redis.exceptions import (
@@ -32,6 +31,7 @@ from redis.exceptions import (
     ResponseError,
     TimeoutError,
 )
+from redis.parsers import CommandsParser
 from redis.retry import Retry
 from redis.utils import str_if_bytes
 from tests.test_pubsub import wait_for_message
diff --git a/tests/test_command_parser.py b/tests/test_command_parser.py
index 6c3ede9..b2a2268 100644
--- a/tests/test_command_parser.py
+++ b/tests/test_command_parser.py
@@ -1,6 +1,6 @@
 import pytest
 
-from redis.commands import CommandsParser
+from redis.parsers import CommandsParser
 
 from .conftest import skip_if_redis_enterprise, skip_if_server_version_lt
 
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index 73eb6e1..333ffdb 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -7,7 +7,8 @@ from unittest import mock
 import pytest
 
 import redis
-from redis.connection import ssl_available, to_bool
+from redis.connection import to_bool
+from redis.utils import SSL_AVAILABLE
 
 from .conftest import _get_client, skip_if_redis_enterprise, skip_if_server_version_lt
 from .test_pubsub import wait_for_message
@@ -425,7 +426,7 @@ class TestConnectionPoolUnixSocketURLParsing:
         assert pool.connection_class == MyConnection
 
 
-@pytest.mark.skipif(not ssl_available, reason="SSL not installed")
+@pytest.mark.skipif(not SSL_AVAILABLE, reason="SSL not installed")
 class TestSSLConnectionURLParsing:
     def test_host(self):
         pool = redis.ConnectionPool.from_url("rediss://my.host")
diff --git a/tests/test_encoding.py b/tests/test_encoding.py
index 2867640..ee52e4c 100644
--- a/tests/test_encoding.py
+++ b/tests/test_encoding.py
@@ -97,10 +97,7 @@ class TestCommandsAreNotEncoded:
 
 
 class TestInvalidUserInput:
-    def test_boolean_fails(self, r):
-        with pytest.raises(redis.DataError):
-            r.set("a", True)
-
+    
     def test_none_fails(self, r):
         with pytest.raises(redis.DataError):
             r.set("a", None)
diff --git a/tests/test_pubsub.py b/tests/test_pubsub.py
index 5d86934..d131ff7 100644
--- a/tests/test_pubsub.py
+++ b/tests/test_pubsub.py
@@ -767,9 +767,9 @@ class TestBaseException:
         assert msg is not None
         # timeout waiting for another message which never arrives
         assert is_connected()
-        with patch("redis.connection.PythonParser.read_response") as mock1:
+        with patch("redis.parsers.RESP2Parser.read_response") as mock1:
             mock1.side_effect = BaseException("boom")
-            with patch("redis.connection.HiredisParser.read_response") as mock2:
+            with patch("redis.parsers.HiredisParser.read_response") as mock2:
                 mock2.side_effect = BaseException("boom")
 
                 with pytest.raises(BaseException):
diff --git a/whitelist.py b/whitelist.py
index 8c9cee3..29cd529 100644
--- a/whitelist.py
+++ b/whitelist.py
@@ -14,6 +14,5 @@ exc_type  # unused variable (/data/repos/redis/redis-py/redis/asyncio/utils.py:2
 exc_value  # unused variable (/data/repos/redis/redis-py/redis/asyncio/utils.py:26)
 traceback  # unused variable (/data/repos/redis/redis-py/redis/asyncio/utils.py:26)
 AsyncConnectionPool  # unused import (//data/repos/redis/redis-py/redis/typing.py:9)
-AsyncEncoder  # unused import (//data/repos/redis/redis-py/redis/typing.py:10)
 AsyncRedis  # unused import (//data/repos/redis/redis-py/redis/commands/core.py:49)
 TargetNodesT  # unused import (//data/repos/redis/redis-py/redis/commands/cluster.py:46)