Repository: redis/redis-py
Pull request #2588 (2023-02-11 08:01:38)

introduce AbstractConnection so that UnixDomainSocketConnection can call super().__init__
### Pull Request check-list

_Please make sure to review and check all of these items:_

- [x] Does `$ tox` pass with this change (including linting)?
- [x] Do the CI tests pass with this change (enable it first in your forked repo and wait for the github action build to finish)?
- [ ] Is the new or changed code fully tested?
- [ ] Is a documentation update included (if this change modifies existing APIs, or introduces new ones)?
- [ ] Is there an example added to the examples folder (if applicable)?
- [ ] Was the change added to CHANGES file?

_NOTE: these things are not required to open a PR and can be done
afterwards / while the PR is open._

### Description of change

**Motivation**: changes to `Connection.__init__` are easily forgotten to be applied also to `UnixDomainSocketConnection.__init__` (for example #2570 and #2377) which leads to issues like #2581.

**Before this MR**

```python
class Connection:
    ...

class UnixDomainSocketConnection(Connection):
    def __init__(...):
        # does not call super().__init__(...)
```

**After this MR**

```python
class AbstractConnection:
    ...

class Connection(AbstractConnection):
    def __init__(...):
        ...
        super().__init__(...)

class UnixDomainSocketConnection(AbstractConnection):
    def __init__(...):
        ...
        super().__init__(...)
```

codecov-commenter: # [Codecov](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Base: **92.29**% // Head: **78.14**% // Decreases project coverage by **`-14.15%`** :warning:
> Coverage data is based on head [(`239c580`)](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) compared to base [(`fd7a79d`)](https://codecov.io/gh/redis/redis-py/commit/fd7a79dca3d10af88a2ef79f81471a63376c081f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).
> Patch coverage: 78.26% of modified lines in pull request are covered.

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           master    #2588       +/-   ##
===========================================
- Coverage   92.29%   78.14%   -14.15%     
===========================================
  Files         115      115               
  Lines       29744    29729       -15     
===========================================
- Hits        27451    23232     -4219     
- Misses       2293     6497     +4204     
```


| [Impacted Files](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [redis/connection.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29ubmVjdGlvbi5weQ==) | `77.52% <78.26%> (-9.27%)` | :arrow_down: |
| [tests/test\_asyncio/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `18.51% <0.00%> (-78.97%)` | :arrow_down: |
| [tests/test\_cluster.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9jbHVzdGVyLnB5) | `18.00% <0.00%> (-78.87%)` | :arrow_down: |
| [redis/asyncio/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `16.96% <0.00%> (-74.72%)` | :arrow_down: |
| [redis/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2x1c3Rlci5weQ==) | `18.77% <0.00%> (-71.66%)` | :arrow_down: |
| [redis/asyncio/parser.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9wYXJzZXIucHk=) | `17.30% <0.00%> (-63.47%)` | :arrow_down: |
| [redis/commands/cluster.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29tbWFuZHMvY2x1c3Rlci5weQ==) | `36.95% <0.00%> (-57.07%)` | :arrow_down: |
| [redis/crc.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY3JjLnB5) | `45.45% <0.00%> (-54.55%)` | :arrow_down: |
| [tests/mocks.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvbW9ja3MucHk=) | `68.96% <0.00%> (-27.59%)` | :arrow_down: |
| [tests/test\_asyncio/test\_connection.py](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbi5weQ==) | `75.79% <0.00%> (-22.30%)` | :arrow_down: |
| ... and [33 more](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/redis/redis-py/pull/2588?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).


Patch:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 8d38cd45..0f9db8fb 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -52,7 +52,7 @@ jobs:
      strategy:
        max-parallel: 15
        matrix:
-         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8']
+         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']
          connection-type: ['hiredis', 'plain']
      env:
@@ -109,7 +109,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7']
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11', 'pypy-3.7', 'pypy-3.8', 'pypy-3.9']
     steps:
       - uses: actions/checkout@v3
       - uses: actions/setup-python@v4
diff --git a/CHANGES b/CHANGES
index e83660d6..3e4eba44 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Use asyncio.timeout() instead of async_timeout.timeout() for python >= 3.11 (#2602)
     * Add test and fix async HiredisParser when reading during a disconnect() (#2349)
     * Use hiredis-py pack_command if available.
     * Support `.unlink()` in ClusterPipeline
diff --git a/README.md b/README.md
index 6b53b42f..e97119a8 100644
--- a/README.md
+++ b/README.md
@@ -13,12 +13,6 @@ The Python interface to the Redis key-value store.
 
 ---------------------------------------------
 
-## Python Notice
-
-redis-py 4.3.x will be the last generation of redis-py to support python 3.6 as it has been [End of Life'd](https://www.python.org/dev/peps/pep-0494/#schedule-last-security-only-release).  Async support was introduced in redis-py 4.2.x thanks to [aioredis](https://github.com/aio-libs/aioredis-py), which necessitates this change. We will continue to maintain 3.6 support as long as possible - but the plan is for redis-py version 4.4+ to officially remove 3.6.
-
----------------------------
-
 ## Installation
 
 Start a redis via docker:
@@ -42,6 +36,18 @@ $ pip install "redis[hiredis]"
 
 Looking for a high-level library to handle object mapping? See [redis-om-python](https://github.com/redis/redis-om-python)!
 
+## Supported Redis Versions
+
+The most recent version of this library supports redis version [5.0](https://github.com/redis/redis/blob/5.0/00-RELEASENOTES), [6.0](https://github.com/redis/redis/blob/6.0/00-RELEASENOTES), [6.2](https://github.com/redis/redis/blob/6.2/00-RELEASENOTES), and [7.0](https://github.com/redis/redis/blob/7.0/00-RELEASENOTES).
+
+The table below higlights version compatibility of the most-recent library versions and redis versions.
+
+| Library version | Supported redis versions |
+|-----------------|-------------------|
+| 3.5.3 | <= 6.2 Family of releases |
+| >= 4.1.0 | Version 5.0 to current |
+
+
 ## Usage
 
 ### Basic Example
diff --git a/docs/examples/connection_examples.ipynb b/docs/examples/connection_examples.ipynb
index a15b4c6c..7f5ac53e 100644
--- a/docs/examples/connection_examples.ipynb
+++ b/docs/examples/connection_examples.ipynb
@@ -222,18 +222,23 @@
     "import json\n",
     "import cachetools.func\n",
     "\n",
-    "sm_client = boto3.client('secretsmanager')\n",
-    "  \n",
-    "def sm_auth_provider(self, secret_id, version_id=None, version_stage='AWSCURRENT'):\n",
-    "    @cachetools.func.ttl_cache(maxsize=128, ttl=24 * 60 * 60) #24h\n",
-    "    def get_sm_user_credentials(secret_id, version_id, version_stage):\n",
-    "        secret = sm_client.get_secret_value(secret_id, version_id)\n",
-    "        return json.loads(secret['SecretString'])\n",
-    "    creds = get_sm_user_credentials(secret_id, version_id, version_stage)\n",
-    "    return creds['username'], creds['password']\n",
+    "class SecretsManagerProvider(redis.CredentialProvider):\n",
+    "    def __init__(self, secret_id, version_id=None, version_stage='AWSCURRENT'):\n",
+    "        self.sm_client = boto3.client('secretsmanager')\n",
+    "        self.secret_id = secret_id\n",
+    "        self.version_id = version_id\n",
+    "        self.version_stage = version_stage\n",
     "\n",
-    "secret_id = \"EXAMPLE1-90ab-cdef-fedc-ba987SECRET1\"\n",
-    "creds_provider = redis.CredentialProvider(supplier=sm_auth_provider, secret_id=secret_id)\n",
+    "    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:\n",
+    "        @cachetools.func.ttl_cache(maxsize=128, ttl=24 * 60 * 60) #24h\n",
+    "        def get_sm_user_credentials(secret_id, version_id, version_stage):\n",
+    "            secret = self.sm_client.get_secret_value(secret_id, version_id)\n",
+    "            return json.loads(secret['SecretString'])\n",
+    "        creds = get_sm_user_credentials(self.secret_id, self.version_id, self.version_stage)\n",
+    "        return creds['username'], creds['password']\n",
+    "\n",
+    "my_secret_id = \"EXAMPLE1-90ab-cdef-fedc-ba987SECRET1\"\n",
+    "creds_provider = SecretsManagerProvider(secret_id=my_secret_id)\n",
     "user_connection = redis.Redis(host=\"localhost\", port=6379, credential_provider=creds_provider)\n",
     "user_connection.ping()"
    ]
@@ -266,19 +271,24 @@
     "import boto3\n",
     "import cachetools.func\n",
     "\n",
-    "ec_client = boto3.client('elasticache')\n",
+    "class ElastiCacheIAMProvider(redis.CredentialProvider):\n",
+    "    def __init__(self, user, endpoint, port=6379, region=\"us-east-1\"):\n",
+    "        self.ec_client = boto3.client('elasticache')\n",
+    "        self.user = user\n",
+    "        self.endpoint = endpoint\n",
+    "        self.port = port\n",
+    "        self.region = region\n",
     "\n",
-    "def iam_auth_provider(self, user, endpoint, port=6379, region=\"us-east-1\"):\n",
-    "    @cachetools.func.ttl_cache(maxsize=128, ttl=15 * 60) # 15m\n",
-    "    def get_iam_auth_token(user, endpoint, port, region):\n",
-    "        return ec_client.generate_iam_auth_token(user, endpoint, port, region)\n",
-    "    iam_auth_token = get_iam_auth_token(endpoint, port, user, region)\n",
-    "    return iam_auth_token\n",
+    "    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:\n",
+    "        @cachetools.func.ttl_cache(maxsize=128, ttl=15 * 60) # 15m\n",
+    "        def get_iam_auth_token(user, endpoint, port, region):\n",
+    "            return self.ec_client.generate_iam_auth_token(user, endpoint, port, region)\n",
+    "        iam_auth_token = get_iam_auth_token(self.endpoint, self.port, self.user, self.region)\n",
+    "        return iam_auth_token\n",
     "\n",
     "username = \"barshaul\"\n",
     "endpoint = \"test-001.use1.cache.amazonaws.com\"\n",
-    "creds_provider = redis.CredentialProvider(supplier=iam_auth_provider, user=username,\n",
-    "                                           endpoint=endpoint)\n",
+    "creds_provider = ElastiCacheIAMProvider(user=username, endpoint=endpoint)\n",
     "user_connection = redis.Redis(host=endpoint, port=6379, credential_provider=creds_provider)\n",
     "user_connection.ping()"
    ]
diff --git a/docs/examples/ssl_connection_examples.ipynb b/docs/examples/ssl_connection_examples.ipynb
index 386e4af4..ab3b4415 100644
--- a/docs/examples/ssl_connection_examples.ipynb
+++ b/docs/examples/ssl_connection_examples.ipynb
@@ -55,6 +55,27 @@
     "url_connection.ping()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "id": "04e70233",
+   "metadata": {},
+   "source": [
+    "## Connecting to a Redis instance using ConnectionPool"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2903de26",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import redis\n",
+    "redis_pool = redis.ConnectionPool(host=\"localhost\", port=6666, connection_class=redis.SSLConnection)\n",
+    "ssl_connection = redis.StrictRedis(connection_pool=redis_pool) \n",
+    "ssl_connection.ping()"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index e77fba30..93db37e4 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -5,6 +5,7 @@ import inspect
 import os
 import socket
 import ssl
+import sys
 import threading
 import weakref
 from itertools import chain
@@ -24,7 +25,11 @@ from typing import (
 )
 from urllib.parse import ParseResult, parse_qs, unquote, urlparse
 
-import async_timeout
+if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
+    from asyncio import timeout as async_timeout
+else:
+    from async_timeout import timeout as async_timeout
+
 
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
@@ -242,7 +247,7 @@ class PythonParser(BaseParser):
         if self._stream is None:
             raise RedisError("Buffer is closed.")
         try:
-            async with async_timeout.timeout(0):
+            async with async_timeout(0):
                 return await self._stream.read(1)
         except asyncio.TimeoutError:
             return False
@@ -267,9 +272,6 @@ class PythonParser(BaseParser):
         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")
@@ -289,22 +291,24 @@ class PythonParser(BaseParser):
             pass
         # int value
         elif byte == b":":
-            response = int(response)
+            return int(response)
         # bulk response
+        elif byte == b"$" and response == b"-1":
+            return None
         elif byte == b"$":
-            length = int(response)
-            if length == -1:
-                return None
-            response = await self._read(length)
+            response = await self._read(int(response))
         # multi-bulk response
+        elif byte == b"*" and response == b"-1":
+            return None
         elif byte == b"*":
-            length = int(response)
-            if length == -1:
-                return None
             response = [
-                (await self._read_response(disable_decoding)) for _ in range(length)
+                (await self._read_response(disable_decoding))
+                for _ in range(int(response))  # noqa
             ]
-        if isinstance(response, bytes) and disable_decoding is False:
+        else:
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        if disable_decoding is False:
             response = self.encoder.decode(response)
         return response
 
@@ -381,7 +385,7 @@ class HiredisParser(BaseParser):
         if self._reader.gets():
             return True
         try:
-            async with async_timeout.timeout(0):
+            async with async_timeout(0):
                 return await self.read_from_socket()
         except asyncio.TimeoutError:
             return False
@@ -636,7 +640,7 @@ class Connection:
 
     async def _connect(self):
         """Create a TCP socket connection"""
-        async with async_timeout.timeout(self.socket_connect_timeout):
+        async with async_timeout(self.socket_connect_timeout):
             reader, writer = await asyncio.open_connection(
                 host=self.host,
                 port=self.port,
@@ -723,7 +727,7 @@ class Connection:
     async def disconnect(self, nowait: bool = False) -> None:
         """Disconnects from the Redis server"""
         try:
-            async with async_timeout.timeout(self.socket_connect_timeout):
+            async with async_timeout(self.socket_connect_timeout):
                 self._parser.on_disconnect()
                 if not self.is_connected:
                     return
@@ -828,7 +832,7 @@ class Connection:
         read_timeout = timeout if timeout is not None else self.socket_timeout
         try:
             if read_timeout is not None:
-                async with async_timeout.timeout(read_timeout):
+                async with async_timeout(read_timeout):
                     response = await self._parser.read_response(
                         disable_decoding=disable_decoding
                     )
@@ -1119,7 +1123,7 @@ class UnixDomainSocketConnection(Connection):  # lgtm [py/missing-call-to-init]
         return pieces
 
     async def _connect(self):
-        async with async_timeout.timeout(self.socket_connect_timeout):
+        async with async_timeout(self.socket_connect_timeout):
             reader, writer = await asyncio.open_unix_connection(path=self.path)
         self._reader = reader
         self._writer = writer
@@ -1590,7 +1594,7 @@ class BlockingConnectionPool(ConnectionPool):
         # self.timeout then raise a ``ConnectionError``.
         connection = None
         try:
-            async with async_timeout.timeout(self.timeout):
+            async with async_timeout(self.timeout):
                 connection = await self.pool.get()
         except (asyncio.QueueEmpty, asyncio.TimeoutError):
             # Note that this is not caught by the redis client and will be
diff --git a/redis/cluster.py b/redis/cluster.py
index d6dc02d4..5e6e7da5 100644
--- a/redis/cluster.py
+++ b/redis/cluster.py
@@ -135,6 +135,7 @@ REDIS_ALLOWED_KEYS = (
     "redis_connect_func",
     "password",
     "port",
+    "queue_class",
     "retry",
     "retry_on_timeout",
     "socket_connect_timeout",
@@ -588,7 +589,6 @@ class RedisCluster(AbstractRedisCluster, RedisClusterCommands):
         self.read_from_replicas = read_from_replicas
         self.reinitialize_counter = 0
         self.reinitialize_steps = reinitialize_steps
-        self.nodes_manager = None
         self.nodes_manager = NodesManager(
             startup_nodes=startup_nodes,
             from_url=from_url,
diff --git a/redis/commands/core.py b/redis/commands/core.py
index 28dab81f..e2cabb85 100644
--- a/redis/commands/core.py
+++ b/redis/commands/core.py
@@ -3357,10 +3357,15 @@ class SetCommands(CommandsProtocol):
 
     def smismember(
         self, name: str, values: List, *args: List
-    ) -> Union[Awaitable[List[bool]], List[bool]]:
+    ) -> Union[
+        Awaitable[List[Union[Literal[0], Literal[1]]]],
+        List[Union[Literal[0], Literal[1]]],
+    ]:
         """
         Return whether each value in ``values`` is a member of the set ``name``
-        as a list of ``bool`` in the order of ``values``
+        as a list of ``int`` in the order of ``values``:
+        - 1 if the value is a member of the set.
+        - 0 if the value is not a member of the set or if key does not exist.
 
         For more information see https://redis.io/commands/smismember
         """
diff --git a/redis/commands/json/commands.py b/redis/commands/json/commands.py
index 7fd40392..c02c47ad 100644
--- a/redis/commands/json/commands.py
+++ b/redis/commands/json/commands.py
@@ -31,8 +31,8 @@ class JSONCommands:
         name: str,
         path: str,
         scalar: int,
-        start: Optional[int] = 0,
-        stop: Optional[int] = -1,
+        start: Optional[int] = None,
+        stop: Optional[int] = None,
     ) -> List[Union[int, None]]:
         """
         Return the index of ``scalar`` in the JSON array under ``path`` at key
@@ -43,9 +43,13 @@ class JSONCommands:
 
         For more information see `JSON.ARRINDEX <https://redis.io/commands/json.arrindex>`_.
         """  # noqa
-        return self.execute_command(
-            "JSON.ARRINDEX", name, str(path), self._encode(scalar), start, stop
-        )
+        pieces = [name, str(path), self._encode(scalar)]
+        if start is not None:
+            pieces.append(start)
+            if stop is not None:
+                pieces.append(stop)
+
+        return self.execute_command("JSON.ARRINDEX", *pieces)
 
     def arrinsert(
         self, name: str, path: str, index: int, *args: List[JsonType]
diff --git a/redis/commands/search/document.py b/redis/commands/search/document.py
index 5b305054..47534ec2 100644
--- a/redis/commands/search/document.py
+++ b/redis/commands/search/document.py
@@ -11,3 +11,7 @@ class Document:
 
     def __repr__(self):
         return f"Document {self.__dict__}"
+
+    def __getitem__(self, item):
+        value = getattr(self, item)
+        return value
diff --git a/redis/connection.py b/redis/connection.py
index d35980c1..faea7683 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -6,6 +6,7 @@ import socket
 import sys
 import threading
 import weakref
+from abc import abstractmethod
 from io import SEEK_END
 from itertools import chain
 from queue import Empty, Full, LifoQueue
@@ -358,9 +359,6 @@ class PythonParser(BaseParser):
 
         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")
@@ -379,23 +377,24 @@ class PythonParser(BaseParser):
             pass
         # int value
         elif byte == b":":
-            response = int(response)
+            return int(response)
         # bulk response
+        elif byte == b"$" and response == b"-1":
+            return None
         elif byte == b"$":
-            length = int(response)
-            if length == -1:
-                return None
-            response = self._buffer.read(length)
+            response = self._buffer.read(int(response))
         # multi-bulk response
+        elif byte == b"*" and response == b"-1":
+            return None
         elif byte == b"*":
-            length = int(response)
-            if length == -1:
-                return None
             response = [
                 self._read_response(disable_decoding=disable_decoding)
-                for i in range(length)
+                for i in range(int(response))
             ]
-        if isinstance(response, bytes) and disable_decoding is False:
+        else:
+            raise InvalidResponse(f"Protocol Error: {raw!r}")
+
+        if disable_decoding is False:
             response = self.encoder.decode(response)
         return response
 
@@ -585,20 +584,13 @@ class PythonRespSerializer:
         return output
 
 
-class Connection:
-    "Manages TCP communication to and from a Redis server"
+class AbstractConnection:
+    "Manages communication to and from a Redis server"
 
     def __init__(
         self,
-        host="localhost",
-        port=6379,
         db=0,
         password=None,
-        socket_timeout=None,
-        socket_connect_timeout=None,
-        socket_keepalive=False,
-        socket_keepalive_options=None,
-        socket_type=0,
         retry_on_timeout=False,
         retry_on_error=SENTINEL,
         encoding="utf-8",
@@ -629,18 +621,11 @@ class Connection:
                 "2. 'credential_provider'"
             )
         self.pid = os.getpid()
-        self.host = host
-        self.port = int(port)
         self.db = db
         self.client_name = client_name
         self.credential_provider = credential_provider
         self.password = password
         self.username = username
-        self.socket_timeout = socket_timeout
-        self.socket_connect_timeout = socket_connect_timeout or socket_timeout
-        self.socket_keepalive = socket_keepalive
-        self.socket_keepalive_options = socket_keepalive_options or {}
-        self.socket_type = socket_type
         self.retry_on_timeout = retry_on_timeout
         if retry_on_error is SENTINEL:
             retry_on_error = []
@@ -673,11 +658,9 @@ class Connection:
         repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()])
         return f"{self.__class__.__name__}<{repr_args}>"
 
+    @abstractmethod
     def repr_pieces(self):
-        pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
-        if self.client_name:
-            pieces.append(("client_name", self.client_name))
-        return pieces
+        pass
 
     def __del__(self):
         try:
@@ -740,75 +723,17 @@ class Connection:
             if callback:
                 callback(self)
 
+    @abstractmethod
     def _connect(self):
-        "Create a TCP socket connection"
-        # we want to mimic what socket.create_connection does to support
-        # ipv4/ipv6, but we want to set options prior to calling
-        # socket.connect()
-        err = None
-        for res in socket.getaddrinfo(
-            self.host, self.port, self.socket_type, socket.SOCK_STREAM
-        ):
-            family, socktype, proto, canonname, socket_address = res
-            sock = None
-            try:
-                sock = socket.socket(family, socktype, proto)
-                # TCP_NODELAY
-                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
-                # TCP_KEEPALIVE
-                if self.socket_keepalive:
-                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-                    for k, v in self.socket_keepalive_options.items():
-                        sock.setsockopt(socket.IPPROTO_TCP, k, v)
-
-                # set the socket_connect_timeout before we connect
-                sock.settimeout(self.socket_connect_timeout)
-
-                # connect
-                sock.connect(socket_address)
-
-                # set the socket_timeout now that we're connected
-                sock.settimeout(self.socket_timeout)
-                return sock
-
-            except OSError as _:
-                err = _
-                if sock is not None:
-                    sock.close()
-
-        if err is not None:
-            raise err
-        raise OSError("socket.getaddrinfo returned an empty list")
+        pass
 
+    @abstractmethod
     def _host_error(self):
-        try:
-            host_error = f"{self.host}:{self.port}"
-        except AttributeError:
-            host_error = "connection"
-
-        return host_error
+        pass
 
+    @abstractmethod
     def _error_message(self, exception):
-        # args for socket.error can either be (errno, "message")
-        # or just "message"
-
-        host_error = self._host_error()
-
-        if len(exception.args) == 1:
-            try:
-                return f"Error connecting to {host_error}. \
-                        {exception.args[0]}."
-            except AttributeError:
-                return f"Connection Error: {exception.args[0]}"
-        else:
-            try:
-                return (
-                    f"Error {exception.args[0]} connecting to "
-                    f"{host_error}. {exception.args[1]}."
-                )
-            except AttributeError:
-                return f"Connection Error: {exception.args[0]}"
+        pass
 
     def on_connect(self):
         "Initialize the connection, authenticate and select a database"
@@ -992,6 +917,101 @@ class Connection:
         return output
 
 
+class Connection(AbstractConnection):
+    "Manages TCP communication to and from a Redis server"
+
+    def __init__(
+        self,
+        host="localhost",
+        port=6379,
+        socket_timeout=None,
+        socket_connect_timeout=None,
+        socket_keepalive=False,
+        socket_keepalive_options=None,
+        socket_type=0,
+        **kwargs,
+    ):
+        self.host = host
+        self.port = int(port)
+        self.socket_timeout = socket_timeout
+        self.socket_connect_timeout = socket_connect_timeout or socket_timeout
+        self.socket_keepalive = socket_keepalive
+        self.socket_keepalive_options = socket_keepalive_options or {}
+        self.socket_type = socket_type
+        super().__init__(**kwargs)
+
+    def repr_pieces(self):
+        pieces = [("host", self.host), ("port", self.port), ("db", self.db)]
+        if self.client_name:
+            pieces.append(("client_name", self.client_name))
+        return pieces
+
+    def _connect(self):
+        "Create a TCP socket connection"
+        # we want to mimic what socket.create_connection does to support
+        # ipv4/ipv6, but we want to set options prior to calling
+        # socket.connect()
+        err = None
+        for res in socket.getaddrinfo(
+            self.host, self.port, self.socket_type, socket.SOCK_STREAM
+        ):
+            family, socktype, proto, canonname, socket_address = res
+            sock = None
+            try:
+                sock = socket.socket(family, socktype, proto)
+                # TCP_NODELAY
+                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+
+                # TCP_KEEPALIVE
+                if self.socket_keepalive:
+                    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+                    for k, v in self.socket_keepalive_options.items():
+                        sock.setsockopt(socket.IPPROTO_TCP, k, v)
+
+                # set the socket_connect_timeout before we connect
+                sock.settimeout(self.socket_connect_timeout)
+
+                # connect
+                sock.connect(socket_address)
+
+                # set the socket_timeout now that we're connected
+                sock.settimeout(self.socket_timeout)
+                return sock
+
+            except OSError as _:
+                err = _
+                if sock is not None:
+                    sock.close()
+
+        if err is not None:
+            raise err
+        raise OSError("socket.getaddrinfo returned an empty list")
+
+    def _host_error(self):
+        return f"{self.host}:{self.port}"
+
+    def _error_message(self, exception):
+        # args for socket.error can either be (errno, "message")
+        # or just "message"
+
+        host_error = self._host_error()
+
+        if len(exception.args) == 1:
+            try:
+                return f"Error connecting to {host_error}. \
+                        {exception.args[0]}."
+            except AttributeError:
+                return f"Connection Error: {exception.args[0]}"
+        else:
+            try:
+                return (
+                    f"Error {exception.args[0]} connecting to "
+                    f"{host_error}. {exception.args[1]}."
+                )
+            except AttributeError:
+                return f"Connection Error: {exception.args[0]}"
+
+
 class SSLConnection(Connection):
     """Manages SSL connections to and from the Redis server(s).
     This class extends the Connection class, adding SSL functionality, and making
@@ -1037,8 +1057,6 @@ class SSLConnection(Connection):
         if not ssl_available:
             raise RedisError("Python wasn't built with SSL support")
 
-        super().__init__(**kwargs)
-
         self.keyfile = ssl_keyfile
         self.certfile = ssl_certfile
         if ssl_cert_reqs is None:
@@ -1064,6 +1082,7 @@ class SSLConnection(Connection):
         self.ssl_validate_ocsp_stapled = ssl_validate_ocsp_stapled
         self.ssl_ocsp_context = ssl_ocsp_context
         self.ssl_ocsp_expected_cert = ssl_ocsp_expected_cert
+        super().__init__(**kwargs)
 
     def _connect(self):
         "Wrap the socket with SSL support"
@@ -1133,77 +1152,12 @@ class SSLConnection(Connection):
         return sslsock
 
 
-class UnixDomainSocketConnection(Connection):
-    def __init__(
-        self,
-        path="",
-        db=0,
-        username=None,
-        password=None,
-        socket_timeout=None,
-        encoding="utf-8",
-        encoding_errors="strict",
-        decode_responses=False,
-        retry_on_timeout=False,
-        retry_on_error=SENTINEL,
-        parser_class=DefaultParser,
-        socket_read_size=65536,
-        health_check_interval=0,
-        client_name=None,
-        retry=None,
-        redis_connect_func=None,
-        credential_provider: Optional[CredentialProvider] = None,
-        command_packer=None,
-    ):
-        """
-        Initialize a new UnixDomainSocketConnection.
-        To specify a retry policy for specific errors, first set
-        `retry_on_error` to a list of the error/s to retry on, then set
-        `retry` to a valid `Retry` object.
-        To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
-        """
-        if (username or password) and credential_provider is not None:
-            raise DataError(
-                "'username' and 'password' cannot be passed along with 'credential_"
-                "provider'. Please provide only one of the following arguments: \n"
-                "1. 'password' and (optional) 'username'\n"
-                "2. 'credential_provider'"
-            )
-        self.pid = os.getpid()
+class UnixDomainSocketConnection(AbstractConnection):
+    "Manages UDS communication to and from a Redis server"
+
+    def __init__(self, path="", **kwargs):
         self.path = path
-        self.db = db
-        self.client_name = client_name
-        self.credential_provider = credential_provider
-        self.password = password
-        self.username = username
-        self.socket_timeout = socket_timeout
-        self.retry_on_timeout = retry_on_timeout
-        if retry_on_error is SENTINEL:
-            retry_on_error = []
-        if retry_on_timeout:
-            # Add TimeoutError to the errors list to retry on
-            retry_on_error.append(TimeoutError)
-        self.retry_on_error = retry_on_error
-        if self.retry_on_error:
-            if retry is None:
-                self.retry = Retry(NoBackoff(), 1)
-            else:
-                # deep-copy the Retry object as it is mutable
-                self.retry = copy.deepcopy(retry)
-            # Update the retry's supported errors with the specified errors
-            self.retry.update_supported_errors(retry_on_error)
-        else:
-            self.retry = Retry(NoBackoff(), 0)
-        self.health_check_interval = health_check_interval
-        self.next_health_check = 0
-        self.redis_connect_func = redis_connect_func
-        self.encoder = Encoder(encoding, encoding_errors, decode_responses)
-        self._sock = None
-        self._socket_read_size = socket_read_size
-        self.set_parser(parser_class)
-        self._connect_callbacks = []
-        self._buffer_cutoff = 6000
-        self._command_packer = self._construct_command_packer(command_packer)
+        super().__init__(**kwargs)
 
     def repr_pieces(self):
         pieces = [("path", self.path), ("db", self.db)]
@@ -1218,15 +1172,21 @@ class UnixDomainSocketConnection(Connection):
         sock.connect(self.path)
         return sock
 
+    def _host_error(self):
+        return self.path
+
     def _error_message(self, exception):
         # args for socket.error can either be (errno, "message")
         # or just "message"
+        host_error = self._host_error()
         if len(exception.args) == 1:
-            return f"Error connecting to unix socket: {self.path}. {exception.args[0]}."
+            return (
+                f"Error connecting to unix socket: {host_error}. {exception.args[0]}."
+            )
         else:
             return (
                 f"Error {exception.args[0]} connecting to unix socket: "
-                f"{self.path}. {exception.args[1]}."
+                f"{host_error}. {exception.args[1]}."
             )
 
 
diff --git a/setup.py b/setup.py
index 060e9da7..ceeeb816 100644
--- a/setup.py
+++ b/setup.py
@@ -34,7 +34,7 @@ setup(
     install_requires=[
         'importlib-metadata >= 1.0; python_version < "3.8"',
         'typing-extensions; python_version<"3.8"',
-        "async-timeout>=4.0.2",
+        'async-timeout>=4.0.2; python_version<"3.11"',
     ],
     classifiers=[
         "Development Status :: 5 - Production/Stable",


Test patch:
diff --git a/tests/test_asyncio/test_json.py b/tests/test_asyncio/test_json.py
index b8854d20..fc530c63 100644
--- a/tests/test_asyncio/test_json.py
+++ b/tests/test_asyncio/test_json.py
@@ -145,9 +145,15 @@ async def test_arrappend(modclient: redis.Redis):
 
 @pytest.mark.redismod
 async def test_arrindex(modclient: redis.Redis):
-    await modclient.json().set("arr", Path.root_path(), [0, 1, 2, 3, 4])
-    assert 1 == await modclient.json().arrindex("arr", Path.root_path(), 1)
-    assert -1 == await modclient.json().arrindex("arr", Path.root_path(), 1, 2)
+    r_path = Path.root_path()
+    await modclient.json().set("arr", r_path, [0, 1, 2, 3, 4])
+    assert 1 == await modclient.json().arrindex("arr", r_path, 1)
+    assert -1 == await modclient.json().arrindex("arr", r_path, 1, 2)
+    assert 4 == await modclient.json().arrindex("arr", r_path, 4)
+    assert 4 == await modclient.json().arrindex("arr", r_path, 4, start=0)
+    assert 4 == await modclient.json().arrindex("arr", r_path, 4, start=0, stop=5000)
+    assert -1 == await modclient.json().arrindex("arr", r_path, 4, start=0, stop=-1)
+    assert -1 == await modclient.json().arrindex("arr", r_path, 4, start=1, stop=3)
 
 
 @pytest.mark.redismod
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index c2a9130e..0df7847e 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -5,7 +5,11 @@ import sys
 from typing import Optional
 from unittest.mock import patch
 
-import async_timeout
+if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
+    from asyncio import timeout as async_timeout
+else:
+    from async_timeout import timeout as async_timeout
+
 import pytest
 import pytest_asyncio
 
@@ -21,7 +25,7 @@ def with_timeout(t):
     def wrapper(corofunc):
         @functools.wraps(corofunc)
         async def run(*args, **kwargs):
-            async with async_timeout.timeout(t):
+            async with async_timeout(t):
                 return await corofunc(*args, **kwargs)
 
         return run
@@ -648,7 +652,7 @@ class TestPubSubReconnect:
 
         async def loop():
             # must make sure the task exits
-            async with async_timeout.timeout(2):
+            async with async_timeout(2):
                 nonlocal interrupt
                 await pubsub.subscribe("foo")
                 while True:
@@ -677,7 +681,7 @@ class TestPubSubReconnect:
 
         task = asyncio.get_running_loop().create_task(loop())
         # get the initial connect message
-        async with async_timeout.timeout(1):
+        async with async_timeout(1):
             message = await messages.get()
         assert message == {
             "channel": b"foo",
@@ -776,7 +780,7 @@ class TestPubSubRun:
             if n == 1:
                 break
             await asyncio.sleep(0.1)
-        async with async_timeout.timeout(0.1):
+        async with async_timeout(0.1):
             message = await messages.get()
         task.cancel()
         # we expect a cancelled error, not the Runtime error
@@ -839,7 +843,7 @@ class TestPubSubAutoReconnect:
         Test that a socket error will cause reconnect
         """
         try:
-            async with async_timeout.timeout(self.timeout):
+            async with async_timeout(self.timeout):
                 await self.mysetup(r, method)
                 # now, disconnect the connection, and wait for it to be re-established
                 async with self.cond:
@@ -868,7 +872,7 @@ class TestPubSubAutoReconnect:
         Test that a manual disconnect() will cause reconnect
         """
         try:
-            async with async_timeout.timeout(self.timeout):
+            async with async_timeout(self.timeout):
                 await self.mysetup(r, method)
                 # now, disconnect the connection, and wait for it to be re-established
                 async with self.cond:
@@ -923,7 +927,7 @@ class TestPubSubAutoReconnect:
     async def loop_step_listen(self):
         # get a single message via listen()
         try:
-            async with async_timeout.timeout(0.1):
+            async with async_timeout(0.1):
                 async for message in self.pubsub.listen():
                     await self.messages.put(message)
                     return True
@@ -947,7 +951,7 @@ class TestBaseException:
         assert pubsub.connection.is_connected
 
         async def get_msg_or_timeout(timeout=0.1):
-            async with async_timeout.timeout(timeout):
+            async with async_timeout(timeout):
                 # blocking method to return messages
                 while True:
                     response = await pubsub.parse_response(block=True)
diff --git a/tests/test_cluster.py b/tests/test_cluster.py
index 1bf57a35..58f9b77d 100644
--- a/tests/test_cluster.py
+++ b/tests/test_cluster.py
@@ -1,6 +1,7 @@
 import binascii
 import datetime
 import warnings
+from queue import LifoQueue, Queue
 from time import sleep
 from unittest.mock import DEFAULT, Mock, call, patch
 
@@ -2511,6 +2512,18 @@ class TestNodesManager:
                 node.redis_connection.connection_pool, connection_pool_class
             )
 
+    @pytest.mark.parametrize("queue_class", [Queue, LifoQueue])
+    def test_allow_custom_queue_class(self, queue_class):
+        rc = get_mocked_redis_client(
+            url="redis://my@DNS.com:7000",
+            cluster_slots=default_cluster_slots,
+            connection_pool_class=BlockingConnectionPool,
+            queue_class=queue_class,
+        )
+
+        for node in rc.nodes_manager.nodes_cache.values():
+            assert node.redis_connection.connection_pool.queue_class == queue_class
+
 
 @pytest.mark.onlycluster
 class TestClusterPubSubObject:
diff --git a/tests/test_json.py b/tests/test_json.py
index a776e9e7..8e8da056 100644
--- a/tests/test_json.py
+++ b/tests/test_json.py
@@ -166,6 +166,11 @@ def test_arrindex(client):
     client.json().set("arr", Path.root_path(), [0, 1, 2, 3, 4])
     assert 1 == client.json().arrindex("arr", Path.root_path(), 1)
     assert -1 == client.json().arrindex("arr", Path.root_path(), 1, 2)
+    assert 4 == client.json().arrindex("arr", Path.root_path(), 4)
+    assert 4 == client.json().arrindex("arr", Path.root_path(), 4, start=0)
+    assert 4 == client.json().arrindex("arr", Path.root_path(), 4, start=0, stop=5000)
+    assert -1 == client.json().arrindex("arr", Path.root_path(), 4, start=0, stop=-1)
+    assert -1 == client.json().arrindex("arr", Path.root_path(), 4, start=1, stop=3)
 
 
 @pytest.mark.redismod
diff --git a/tests/test_search.py b/tests/test_search.py
index 57d4338e..7a242815 100644
--- a/tests/test_search.py
+++ b/tests/test_search.py
@@ -140,7 +140,9 @@ def test_client(client):
 
     for doc in res.docs:
         assert doc.id
+        assert doc["id"]
         assert doc.play == "Henry IV"
+        assert doc["play"] == "Henry IV"
         assert len(doc.txt) > 0
 
     # test no content