Repository: redis/redis-py
Pull request #2898 (2023-08-17 15:11:38)

Provide aclose() / close() for classes requiring lifetime management
### 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)?
- [x] Is the new or changed code fully tested?
- [x] 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)?
- [x] 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

The canonical name for an asynchronous `close()` method in async python is `aclose()`.  Providing this method allows support for standard programming patterns, including the use of the `aclosing` context manager.  Conversely, a `close()` method should be _synchronous_, for the same reason.

This PR:

- renames `close()` to  `aclose()` method to several async classes, keeping the old `close()` for backwards compatibility.
- renames a `reset()` to `aclose()`, keeping `close()` and `reset()` for compatibility in the `asyncio.PubSub` class`.
- Adds an `aclose()` method to the async `Pipeline` class, and a `close()` method to the `Pipeline` class.
- Adds an `aclose()` method to the async `ConnectionPool` class and the `close()` method to the sync `ConnectionPool`.

Having `close()` / `aclose()` allows the use of the `closing` / `aclosing` context managers.  In particular, they can now be used to manage a `ConnectionPool` instance.

**Note:** We _keep_ the old asynchronous `close()` methods as alternatives, for backwards compatibility.  Maybe we should actively deprecate them?

chayim: @kristjanvalur You mentioned deprecating in your PR (and thank you). I'm 100% on board for marking as deprecated the old functions - that way that can be safely removed in _future version unknown_. Can you?

I might argue that we (@dvora-h) should define in our CONTRIBUTING.md or elsewhere how and when we'll deprecate an interface
chayim: @dvora-h WDYT? Want to take it from here?
kristjanvalur: > I'm 100% on board for marking as deprecated the old functions - that way that can be safely removed in _future version unknown_. Can you?

Sure, but what is the proper way do do that?
dvora-h: @kristjanvalur Yes, we have a [deprecated_function](https://github.com/redis/redis-py/blob/ba186d210f4f1e5d16fb27a1ee0afcb930f16a8d/redis/utils.py#L113) decorator. you can add it to the function.
codecov-commenter: ## [Codecov](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) Report
Patch coverage: **`65.12%`** and project coverage change: **`-14.16%`** :warning:
> Comparison is base [(`509c77c`)](https://app.codecov.io/gh/redis/redis-py/commit/509c77c5263188fad239b07c80ecc5b0e8372647?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 91.35% compared to head [(`79de021`)](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) 77.20%.

:exclamation: Your organization needs to install the [Codecov GitHub app](https://github.com/apps/codecov/installations/select_target) to enable full functionality.

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


```diff
@@             Coverage Diff             @@
##           master    #2898       +/-   ##
===========================================
- Coverage   91.35%   77.20%   -14.16%     
===========================================
  Files         126      126               
  Lines       32335    32457      +122     
===========================================
- Hits        29540    25057     -4483     
- Misses       2795     7400     +4605     
```


| [Files Changed](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | Coverage Δ | |
|---|---|---|
| [tests/test\_asyncio/test\_cluster.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY2x1c3Rlci5weQ==) | `18.06% <7.50%> (-78.88%)` | :arrow_down: |
| [tests/test\_asyncio/compat.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL2NvbXBhdC5weQ==) | `47.05% <22.22%> (-27.95%)` | :arrow_down: |
| [redis/asyncio/cluster.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbHVzdGVyLnB5) | `17.22% <33.33%> (-74.45%)` | :arrow_down: |
| [tests/test\_asyncio/conftest.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL2NvbmZ0ZXN0LnB5) | `80.55% <50.00%> (-4.87%)` | :arrow_down: |
| [tests/test\_asyncio/test\_connection\_pool.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfY29ubmVjdGlvbl9wb29sLnB5) | `92.02% <54.54%> (+0.07%)` | :arrow_up: |
| [tests/test\_asyncio/test\_pubsub.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-dGVzdHMvdGVzdF9hc3luY2lvL3Rlc3RfcHVic3ViLnB5) | `91.71% <87.23%> (-0.57%)` | :arrow_down: |
| [redis/asyncio/client.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jbGllbnQucHk=) | `92.27% <89.47%> (+0.11%)` | :arrow_up: |
| [redis/asyncio/connection.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvYXN5bmNpby9jb25uZWN0aW9uLnB5) | `82.22% <100.00%> (-1.49%)` | :arrow_down: |
| [redis/client.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY2xpZW50LnB5) | `91.52% <100.00%> (-0.29%)` | :arrow_down: |
| [redis/connection.py](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis#diff-cmVkaXMvY29ubmVjdGlvbi5weQ==) | `78.92% <100.00%> (-3.18%)` | :arrow_down: |
| ... and [4 more](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis) | |

... and [27 files with indirect coverage changes](https://app.codecov.io/gh/redis/redis-py/pull/2898/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/redis/redis-py/pull/2898?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=redis).

dvora-h: @kristjanvalur conflicts... 🙂
dvora-h: @kristjanvalur The CI failed
kristjanvalur: Bummer. Will investigate in a bit 
kristjanvalur: Oops, looks like I pushed the wrong branch today!  Fixing.

Patch:
diff --git a/CHANGES b/CHANGES
index 7b3b4c5a..8cfc47db 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,4 @@
+    * Add 'aclose()' methods to async classes, deprecate async close().
     * Fix #2831, add auto_close_connection_pool=True arg to asyncio.Redis.from_url()
     * Fix incorrect redis.asyncio.Cluster type hint for `retry_on_error`
     * Fix dead weakref in sentinel connection causing ReferenceError (#2767)
diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index a8ee8faf..5eab4db1 100644
--- a/docs/examples/asyncio_examples.ipynb
+++ b/docs/examples/asyncio_examples.ipynb
@@ -15,7 +15,7 @@
     "\n",
     "## Connecting and Disconnecting\n",
     "\n",
-    "Utilizing asyncio Redis requires an explicit disconnect of the connection since there is no asyncio deconstructor magic method. By default, a connection pool is created on `redis.Redis()` and attached to this `Redis` instance. The connection pool closes automatically on the call to `Redis.close` which disconnects all connections."
+    "Utilizing asyncio Redis requires an explicit disconnect of the connection since there is no asyncio deconstructor magic method. By default, a connection pool is created on `redis.Redis()` and attached to this `Redis` instance. The connection pool closes automatically on the call to `Redis.aclose` which disconnects all connections."
    ]
   },
   {
@@ -39,9 +39,9 @@
    "source": [
     "import redis.asyncio as redis\n",
     "\n",
-    "connection = redis.Redis()\n",
-    "print(f\"Ping successful: {await connection.ping()}\")\n",
-    "await connection.close()"
+    "client = redis.Redis()\n",
+    "print(f\"Ping successful: {await client.ping()}\")\n",
+    "await client.aclose()"
    ]
   },
   {
@@ -60,8 +60,8 @@
     "import redis.asyncio as redis\n",
     "\n",
     "pool = redis.ConnectionPool.from_url(\"redis://localhost\")\n",
-    "connection = redis.Redis.from_pool(pool)\n",
-    "await connection.close()"
+    "client = redis.Redis.from_pool(pool)\n",
+    "await client.close()"
    ]
   },
   {
@@ -91,11 +91,11 @@
     "import redis.asyncio as redis\n",
     "\n",
     "pool = redis.ConnectionPool.from_url(\"redis://localhost\")\n",
-    "connection1 = redis.Redis(connection_pool=pool)\n",
-    "connection2 = redis.Redis(connection_pool=pool)\n",
-    "await connection1.close()\n",
-    "await connection2.close()\n",
-    "await pool.disconnect()"
+    "client1 = redis.Redis(connection_pool=pool)\n",
+    "client2 = redis.Redis(connection_pool=pool)\n",
+    "await client1.aclose()\n",
+    "await client2.aclose()\n",
+    "await pool.aclose()"
    ]
   },
   {
@@ -113,9 +113,9 @@
    "source": [
     "import redis.asyncio as redis\n",
     "\n",
-    "connection = redis.Redis(protocol=3)\n",
-    "await connection.close()\n",
-    "await connection.ping()"
+    "client = redis.Redis(protocol=3)\n",
+    "await client.aclose()\n",
+    "await client.ping()"
    ]
   },
   {
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 122469a1..c340d851 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -14,7 +14,6 @@ from typing import (
     List,
     Mapping,
     MutableMapping,
-    NoReturn,
     Optional,
     Set,
     Tuple,
@@ -65,6 +64,7 @@ from redis.typing import ChannelT, EncodableT, KeyT
 from redis.utils import (
     HIREDIS_AVAILABLE,
     _set_info_logger,
+    deprecated_function,
     get_lib_version,
     safe_str,
     str_if_bytes,
@@ -527,7 +527,7 @@ class Redis(
         return await self.initialize()
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.close()
+        await self.aclose()
 
     _DEL_MESSAGE = "Unclosed Redis client"
 
@@ -539,7 +539,7 @@ class Redis(
             context = {"client": self, "message": self._DEL_MESSAGE}
             asyncio.get_running_loop().call_exception_handler(context)
 
-    async def close(self, close_connection_pool: Optional[bool] = None) -> None:
+    async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
         """
         Closes Redis client connection
 
@@ -557,6 +557,13 @@ class Redis(
         ):
             await self.connection_pool.disconnect()
 
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self, close_connection_pool: Optional[bool] = None) -> None:
+        """
+        Alias for aclose(), for backwards compatibility
+        """
+        await self.aclose(close_connection_pool)
+
     async def _send_command_parse_response(self, conn, command_name, *args, **options):
         """
         Send a command and parse the response
@@ -764,13 +771,18 @@ class PubSub:
         return self
 
     async def __aexit__(self, exc_type, exc_value, traceback):
-        await self.reset()
+        await self.aclose()
 
     def __del__(self):
         if self.connection:
             self.connection.clear_connect_callbacks()
 
-    async def reset(self):
+    async def aclose(self):
+        # In case a connection property does not yet exist
+        # (due to a crash earlier in the Redis() constructor), return
+        # immediately as there is nothing to clean-up.
+        if not hasattr(self, "connection"):
+            return
         async with self._lock:
             if self.connection:
                 await self.connection.disconnect()
@@ -782,13 +794,15 @@ class PubSub:
             self.patterns = {}
             self.pending_unsubscribe_patterns = set()
 
-    def close(self) -> Awaitable[NoReturn]:
-        # In case a connection property does not yet exist
-        # (due to a crash earlier in the Redis() constructor), return
-        # immediately as there is nothing to clean-up.
-        if not hasattr(self, "connection"):
-            return
-        return self.reset()
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self) -> None:
+        """Alias for aclose(), for backwards compatibility"""
+        await self.aclose()
+
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="reset")
+    async def reset(self) -> None:
+        """Alias for aclose(), for backwards compatibility"""
+        await self.aclose()
 
     async def on_connect(self, connection: Connection):
         """Re-subscribe to any channels and patterns previously subscribed to"""
@@ -1232,6 +1246,10 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
             await self.connection_pool.release(self.connection)
             self.connection = None
 
+    async def aclose(self) -> None:
+        """Alias for reset(), a standard method name for cleanup"""
+        await self.reset()
+
     def multi(self):
         """
         Start a transactional block of the pipeline after WATCH commands
@@ -1264,14 +1282,14 @@ class Pipeline(Redis):  # lgtm [py/init-calls-subclass]
         # valid since this connection has died. raise a WatchError, which
         # indicates the user should retry this transaction.
         if self.watching:
-            await self.reset()
+            await self.aclose()
             raise WatchError(
                 "A ConnectionError occurred on while watching one or more keys"
             )
         # if retry_on_timeout is not set, or the error is not
         # a TimeoutError, raise it
         if not (conn.retry_on_timeout and isinstance(error, TimeoutError)):
-            await self.reset()
+            await self.aclose()
             raise
 
     async def immediate_execute_command(self, *args, **options):
diff --git a/redis/asyncio/cluster.py b/redis/asyncio/cluster.py
index 65e1bd02..f4f03158 100644
--- a/redis/asyncio/cluster.py
+++ b/redis/asyncio/cluster.py
@@ -62,7 +62,13 @@ from redis.exceptions import (
     TryAgainError,
 )
 from redis.typing import AnyKeyT, EncodableT, KeyT
-from redis.utils import dict_merge, get_lib_version, safe_str, str_if_bytes
+from redis.utils import (
+    deprecated_function,
+    dict_merge,
+    get_lib_version,
+    safe_str,
+    str_if_bytes,
+)
 
 TargetNodesT = TypeVar(
     "TargetNodesT", str, "ClusterNode", List["ClusterNode"], Dict[Any, "ClusterNode"]
@@ -395,12 +401,12 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                         )
                         self._initialize = False
                     except BaseException:
-                        await self.nodes_manager.close()
-                        await self.nodes_manager.close("startup_nodes")
+                        await self.nodes_manager.aclose()
+                        await self.nodes_manager.aclose("startup_nodes")
                         raise
         return self
 
-    async def close(self) -> None:
+    async def aclose(self) -> None:
         """Close all connections & client if initialized."""
         if not self._initialize:
             if not self._lock:
@@ -408,14 +414,19 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
             async with self._lock:
                 if not self._initialize:
                     self._initialize = True
-                    await self.nodes_manager.close()
-                    await self.nodes_manager.close("startup_nodes")
+                    await self.nodes_manager.aclose()
+                    await self.nodes_manager.aclose("startup_nodes")
+
+    @deprecated_function(version="5.0.0", reason="Use aclose() instead", name="close")
+    async def close(self) -> None:
+        """alias for aclose() for backwards compatibility"""
+        await self.aclose()
 
     async def __aenter__(self) -> "RedisCluster":
         return await self.initialize()
 
     async def __aexit__(self, exc_type: None, exc_value: None, traceback: None) -> None:
-        await self.close()
+        await self.aclose()
 
     def __await__(self) -> Generator[Any, None, "RedisCluster"]:
         return self.initialize().__await__()
@@ -767,13 +778,13 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                 self.nodes_manager.startup_nodes.pop(target_node.name, None)
                 # Hard force of reinitialize of the node/slots setup
                 # and try again with the new setup
-                await self.close()
+                await self.aclose()
                 raise
             except ClusterDownError:
                 # ClusterDownError can occur during a failover and to get
                 # self-healed, we will try to reinitialize the cluster layout
                 # and retry executing the command
-                await self.close()
+                await self.aclose()
                 await asyncio.sleep(0.25)
                 raise
             except MovedError as e:
@@ -790,7 +801,7 @@ class RedisCluster(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterCommand
                     self.reinitialize_steps
                     and self.reinitialize_counter % self.reinitialize_steps == 0
                 ):
-                    await self.close()
+                    await self.aclose()
                     # Reset the counter
                     self.reinitialize_counter = 0
                 else:
@@ -1323,7 +1334,7 @@ class NodesManager:
         # If initialize was called after a MovedError, clear it
         self._moved_exception = None
 
-    async def close(self, attr: str = "nodes_cache") -> None:
+    async def aclose(self, attr: str = "nodes_cache") -> None:
         self.default_node = None
         await asyncio.gather(
             *(
@@ -1471,7 +1482,7 @@ class ClusterPipeline(AbstractRedis, AbstractRedisCluster, AsyncRedisClusterComm
                     if type(e) in self.__class__.ERRORS_ALLOW_RETRY:
                         # Try again with the new cluster setup.
                         exception = e
-                        await self._client.close()
+                        await self._client.aclose()
                         await asyncio.sleep(0.25)
                     else:
                         # All other errors should be raised.
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
index 4f20e6ec..71d0e920 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1095,6 +1095,10 @@ class ConnectionPool:
         if exc:
             raise exc
 
+    async def aclose(self) -> None:
+        """Close the pool, disconnecting all connections"""
+        await self.disconnect()
+
     def set_retry(self, retry: "Retry") -> None:
         for conn in self._available_connections:
             conn.retry = retry
diff --git a/redis/client.py b/redis/client.py
index d285e1ca..1e1ff576 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -1217,6 +1217,10 @@ class Pipeline(Redis):
             self.connection_pool.release(self.connection)
             self.connection = None
 
+    def close(self):
+        """Close the pipeline"""
+        self.reset()
+
     def multi(self):
         """
         Start a transactional block of the pipeline after WATCH commands
diff --git a/redis/connection.py b/redis/connection.py
index c6a22aae..45ecd2a3 100644
--- a/redis/connection.py
+++ b/redis/connection.py
@@ -1154,6 +1154,10 @@ class ConnectionPool:
             for connection in connections:
                 connection.disconnect()
 
+    def close(self) -> None:
+        """Close the pool, disconnecting all connections"""
+        self.disconnect()
+
     def set_retry(self, retry: "Retry") -> None:
         self.connection_kwargs.update({"retry": retry})
         for conn in self._available_connections:


Test patch:
diff --git a/tests/test_asyncio/compat.py b/tests/test_asyncio/compat.py
index 5edcd4ae..4a9778b7 100644
--- a/tests/test_asyncio/compat.py
+++ b/tests/test_asyncio/compat.py
@@ -6,6 +6,18 @@ try:
 except AttributeError:
     import mock
 
+try:
+    from contextlib import aclosing
+except ImportError:
+    import contextlib
+
+    @contextlib.asynccontextmanager
+    async def aclosing(thing):
+        try:
+            yield thing
+        finally:
+            await thing.aclose()
+
 
 def create_task(coroutine):
     return asyncio.create_task(coroutine)
diff --git a/tests/test_asyncio/conftest.py b/tests/test_asyncio/conftest.py
index 10ab4732..5d9e0b4f 100644
--- a/tests/test_asyncio/conftest.py
+++ b/tests/test_asyncio/conftest.py
@@ -100,7 +100,7 @@ async def create_redis(request):
                         # handle cases where a test disconnected a client
                         # just manually retry the flushdb
                         await client.flushdb()
-                await client.close()
+                await client.aclose()
                 await client.connection_pool.disconnect()
             else:
                 if flushdb:
@@ -110,7 +110,7 @@ async def create_redis(request):
                         # handle cases where a test disconnected a client
                         # just manually retry the flushdb
                         await client.flushdb(target_nodes="primaries")
-                await client.close()
+                await client.aclose()
 
         teardown_clients.append(teardown)
         return client
diff --git a/tests/test_asyncio/test_cluster.py b/tests/test_asyncio/test_cluster.py
index 332101ed..e6cf2e4c 100644
--- a/tests/test_asyncio/test_cluster.py
+++ b/tests/test_asyncio/test_cluster.py
@@ -38,7 +38,7 @@ from tests.conftest import (
 )
 
 from ..ssl_utils import get_ssl_filename
-from .compat import mock
+from .compat import aclosing, mock
 
 pytestmark = pytest.mark.onlycluster
 
@@ -270,7 +270,38 @@ class TestRedisClusterObj:
         cluster = await get_mocked_redis_client(host=default_host, port=default_port)
         assert cluster.get_node(host=default_host, port=default_port) is not None
 
-        await cluster.close()
+        await cluster.aclose()
+
+    async def test_aclosing(self) -> None:
+        cluster = await get_mocked_redis_client(host=default_host, port=default_port)
+        called = 0
+
+        async def mock_aclose():
+            nonlocal called
+            called += 1
+
+        with mock.patch.object(cluster, "aclose", mock_aclose):
+            async with aclosing(cluster):
+                pass
+            assert called == 1
+        await cluster.aclose()
+
+    async def test_close_is_aclose(self) -> None:
+        """
+        Test that it is possible to use host & port arguments as startup node
+        args
+        """
+        cluster = await get_mocked_redis_client(host=default_host, port=default_port)
+        called = 0
+
+        async def mock_aclose():
+            nonlocal called
+            called += 1
+
+        with mock.patch.object(cluster, "aclose", mock_aclose):
+            await cluster.close()
+            assert called == 1
+        await cluster.aclose()
 
     async def test_startup_nodes(self) -> None:
         """
@@ -289,7 +320,7 @@ class TestRedisClusterObj:
             and cluster.get_node(host=default_host, port=port_2) is not None
         )
 
-        await cluster.close()
+        await cluster.aclose()
 
         startup_node = ClusterNode("127.0.0.1", 16379)
         async with RedisCluster(startup_nodes=[startup_node], client_name="test") as rc:
@@ -417,7 +448,7 @@ class TestRedisClusterObj:
                     )
                 )
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_execute_command_errors(self, r: RedisCluster) -> None:
         """
@@ -461,7 +492,7 @@ class TestRedisClusterObj:
             conn = primary._free.pop()
             assert conn.read_response.called is not True
 
-        await r.close()
+        await r.aclose()
 
     async def test_execute_command_node_flag_all_nodes(self, r: RedisCluster) -> None:
         """
@@ -690,7 +721,7 @@ class TestRedisClusterObj:
                 await read_cluster.get("foo")
                 mocks["send_command"].assert_has_calls([mock.call("READONLY")])
 
-                await read_cluster.close()
+                await read_cluster.aclose()
 
     async def test_keyslot(self, r: RedisCluster) -> None:
         """
@@ -762,7 +793,7 @@ class TestRedisClusterObj:
                 await rc.get("bar")
                 assert execute_command.failed_calls == rc.cluster_error_retry_attempts
 
-            await rc.close()
+            await rc.aclose()
 
     async def test_set_default_node_success(self, r: RedisCluster) -> None:
         """
@@ -843,7 +874,7 @@ class TestRedisClusterObj:
                 *(rc.echo("i", target_nodes=RedisCluster.ALL_NODES) for i in range(100))
             )
         )
-        await rc.close()
+        await rc.aclose()
 
     def test_replace_cluster_node(self, r: RedisCluster) -> None:
         prev_default_node = r.get_default_node()
@@ -901,7 +932,7 @@ class TestRedisClusterObj:
                 assert await r.set("byte_string", b"giraffe")
                 assert await r.get("byte_string") == b"giraffe"
             finally:
-                await r.close()
+                await r.aclose()
         finally:
             await asyncio.gather(*[p.aclose() for p in proxies])
 
@@ -1002,7 +1033,7 @@ class TestClusterRedisCommands:
         url = request.config.getoption("--redis-url")
         r = RedisCluster.from_url(url)
         assert 0 == await r.exists("a", "b", "c")
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_cluster_myid(self, r: RedisCluster) -> None:
@@ -1065,7 +1096,7 @@ class TestClusterRedisCommands:
         assert node0._free.pop().read_response.called
         assert node1._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_server_version_lt("7.0.0")
     @skip_if_redis_enterprise()
@@ -1076,7 +1107,7 @@ class TestClusterRedisCommands:
         await r.cluster_addslots(node, 1, 2, 3, 4, 5)
         assert await r.cluster_delslotsrange(1, 5)
         assert node._free.pop().read_response.called
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_cluster_failover(self, r: RedisCluster) -> None:
@@ -1286,7 +1317,7 @@ class TestClusterRedisCommands:
         for replica in r.get_replicas():
             assert replica._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_readwrite(self) -> None:
@@ -1299,7 +1330,7 @@ class TestClusterRedisCommands:
         for replica in r.get_replicas():
             assert replica._free.pop().read_response.called
 
-        await r.close()
+        await r.aclose()
 
     @skip_if_redis_enterprise()
     async def test_bgsave(self, r: RedisCluster) -> None:
@@ -1524,7 +1555,7 @@ class TestClusterRedisCommands:
         ]
         assert len(clients) == 1
         assert clients[0].get("name") == "redis-py-c1"
-        await r2.close()
+        await r2.aclose()
 
     @skip_if_server_version_lt("2.6.0")
     async def test_cluster_bitop_not_empty_string(self, r: RedisCluster) -> None:
@@ -2302,7 +2333,7 @@ class TestClusterRedisCommands:
 
         await r.acl_deluser(username, target_nodes="primaries")
 
-        await user_client.close()
+        await user_client.aclose()
 
 
 class TestNodesManager:
@@ -2359,7 +2390,7 @@ class TestNodesManager:
                 cluster_slots=cluster_slots,
                 require_full_coverage=True,
             )
-            await rc.close()
+            await rc.aclose()
         assert str(ex.value).startswith(
             "All slots are not covered after query all startup_nodes."
         )
@@ -2385,7 +2416,7 @@ class TestNodesManager:
 
         assert 5460 not in rc.nodes_manager.slots_cache
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_slots_cache(self) -> None:
         """
@@ -2416,7 +2447,7 @@ class TestNodesManager:
 
         assert len(n_manager.nodes_cache) == 6
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_slots_cache_cluster_mode_disabled(self) -> None:
         """
@@ -2427,7 +2458,7 @@ class TestNodesManager:
             rc = await get_mocked_redis_client(
                 host=default_host, port=default_port, cluster_enabled=False
             )
-            await rc.close()
+            await rc.aclose()
         assert "Cluster mode is not enabled on this node" in str(e.value)
 
     async def test_empty_startup_nodes(self) -> None:
@@ -2514,7 +2545,7 @@ class TestNodesManager:
         for i in range(0, REDIS_CLUSTER_HASH_SLOTS):
             assert n.slots_cache[i] == [n_node]
 
-        await rc.close()
+        await rc.aclose()
 
     async def test_init_with_down_node(self) -> None:
         """
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index a954a40d..28e6b0d9 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -302,7 +302,25 @@ async def test_pool_auto_close(request, from_url):
 
     r1 = await get_redis_connection()
     assert r1.auto_close_connection_pool is True
-    await r1.close()
+    await r1.aclose()
+
+
+async def test_close_is_aclose(request):
+    """Verify close() calls aclose()"""
+    calls = 0
+
+    async def mock_aclose(self):
+        nonlocal calls
+        calls += 1
+
+    url: str = request.config.getoption("--redis-url")
+    r1 = await Redis.from_url(url)
+    with patch.object(r1, "aclose", mock_aclose):
+        await r1.close()
+        assert calls == 1
+
+    with pytest.deprecated_call():
+        await r1.close()
 
 
 async def test_pool_from_url_deprecation(request):
@@ -326,7 +344,7 @@ async def test_pool_auto_close_disable(request):
     r1 = await get_redis_connection()
     assert r1.auto_close_connection_pool is False
     await r1.connection_pool.disconnect()
-    await r1.close()
+    await r1.aclose()
 
 
 @pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
diff --git a/tests/test_asyncio/test_connection_pool.py b/tests/test_asyncio/test_connection_pool.py
index 999c0337..c93fa91a 100644
--- a/tests/test_asyncio/test_connection_pool.py
+++ b/tests/test_asyncio/test_connection_pool.py
@@ -7,7 +7,7 @@ import redis.asyncio as redis
 from redis.asyncio.connection import Connection, to_bool
 from tests.conftest import skip_if_redis_enterprise, skip_if_server_version_lt
 
-from .compat import mock
+from .compat import aclosing, mock
 from .conftest import asynccontextmanager
 from .test_pubsub import wait_for_message
 
@@ -42,7 +42,7 @@ class TestRedisAutoReleaseConnectionPool:
         new_conn = await self.create_two_conn(r)
         assert new_conn != r.connection
         assert self.get_total_connected_connections(r.connection_pool) == 2
-        await r.close()
+        await r.aclose()
         assert self.has_no_connected_connections(r.connection_pool)
 
     async def test_do_not_auto_disconnect_redis_created_pool(self, r2: redis.Redis):
@@ -52,7 +52,7 @@ class TestRedisAutoReleaseConnectionPool:
         )
         new_conn = await self.create_two_conn(r2)
         assert self.get_total_connected_connections(r2.connection_pool) == 2
-        await r2.close()
+        await r2.aclose()
         assert r2.connection_pool._in_use_connections == {new_conn}
         assert new_conn.is_connected
         assert len(r2.connection_pool._available_connections) == 1
@@ -61,7 +61,7 @@ class TestRedisAutoReleaseConnectionPool:
     async def test_auto_release_override_true_manual_created_pool(self, r: redis.Redis):
         assert r.auto_close_connection_pool is True, "This is from the class fixture"
         await self.create_two_conn(r)
-        await r.close()
+        await r.aclose()
         assert self.get_total_connected_connections(r.connection_pool) == 2, (
             "The connection pool should not be disconnected as a manually created "
             "connection pool was passed in in conftest.py"
@@ -72,7 +72,7 @@ class TestRedisAutoReleaseConnectionPool:
     async def test_close_override(self, r: redis.Redis, auto_close_conn_pool):
         r.auto_close_connection_pool = auto_close_conn_pool
         await self.create_two_conn(r)
-        await r.close(close_connection_pool=True)
+        await r.aclose(close_connection_pool=True)
         assert self.has_no_connected_connections(r.connection_pool)
 
     @pytest.mark.parametrize("auto_close_conn_pool", [True, False])
@@ -81,7 +81,7 @@ class TestRedisAutoReleaseConnectionPool:
     ):
         r.auto_close_connection_pool = auto_close_conn_pool
         new_conn = await self.create_two_conn(r)
-        await r.close(close_connection_pool=False)
+        await r.aclose(close_connection_pool=False)
         assert not self.has_no_connected_connections(r.connection_pool)
         assert r.connection_pool._in_use_connections == {new_conn}
         assert r.connection_pool._available_connections[0].is_connected
@@ -135,6 +135,16 @@ class TestConnectionPool:
             assert isinstance(connection, DummyConnection)
             assert connection.kwargs == connection_kwargs
 
+    async def test_aclosing(self):
+        connection_kwargs = {"foo": "bar", "biz": "baz"}
+        pool = redis.ConnectionPool(
+            connection_class=DummyConnection,
+            max_connections=None,
+            **connection_kwargs,
+        )
+        async with aclosing(pool):
+            pass
+
     async def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host[0]}
         async with self.get_pool(connection_kwargs=connection_kwargs) as pool:
diff --git a/tests/test_asyncio/test_pipeline.py b/tests/test_asyncio/test_pipeline.py
index edd2f6d1..0fa12047 100644
--- a/tests/test_asyncio/test_pipeline.py
+++ b/tests/test_asyncio/test_pipeline.py
@@ -2,6 +2,7 @@ import pytest
 import redis
 from tests.conftest import skip_if_server_version_lt
 
+from .compat import aclosing, mock
 from .conftest import wait_for_command
 
 
@@ -286,6 +287,24 @@ class TestPipeline:
             assert unwatch_command is not None
             assert unwatch_command["command"] == "UNWATCH"
 
+    @pytest.mark.onlynoncluster
+    async def test_aclose_is_reset(self, r):
+        async with r.pipeline() as pipe:
+            called = 0
+
+            async def mock_reset():
+                nonlocal called
+                called += 1
+
+            with mock.patch.object(pipe, "reset", mock_reset):
+                await pipe.aclose()
+                assert called == 1
+
+    @pytest.mark.onlynoncluster
+    async def test_aclosing(self, r):
+        async with aclosing(r.pipeline()):
+            pass
+
     @pytest.mark.onlynoncluster
     async def test_transaction_callable(self, r):
         await r.set("a", 1)
diff --git a/tests/test_asyncio/test_pubsub.py b/tests/test_asyncio/test_pubsub.py
index 85857658..8fef34d8 100644
--- a/tests/test_asyncio/test_pubsub.py
+++ b/tests/test_asyncio/test_pubsub.py
@@ -20,7 +20,7 @@ from redis.typing import EncodableT
 from redis.utils import HIREDIS_AVAILABLE
 from tests.conftest import get_protocol_version, skip_if_server_version_lt
 
-from .compat import create_task, mock
+from .compat import aclosing, create_task, mock
 
 
 def with_timeout(t):
@@ -84,9 +84,8 @@ def make_subscribe_test_data(pubsub, type):
 
 @pytest_asyncio.fixture()
 async def pubsub(r: redis.Redis):
-    p = r.pubsub()
-    yield p
-    await p.close()
+    async with r.pubsub() as p:
+        yield p
 
 
 @pytest.mark.onlynoncluster
@@ -217,6 +216,46 @@ class TestPubSubSubscribeUnsubscribe:
         kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_subscribed_property(**kwargs)
 
+    async def test_aclosing(self, r: redis.Redis):
+        p = r.pubsub()
+        async with aclosing(p):
+            assert p.subscribed is False
+            await p.subscribe("foo")
+            assert p.subscribed is True
+        assert p.subscribed is False
+
+    async def test_context_manager(self, r: redis.Redis):
+        p = r.pubsub()
+        async with p:
+            assert p.subscribed is False
+            await p.subscribe("foo")
+            assert p.subscribed is True
+        assert p.subscribed is False
+
+    async def test_close_is_aclose(self, r: redis.Redis):
+        """
+        Test backwards compatible close method
+        """
+        p = r.pubsub()
+        assert p.subscribed is False
+        await p.subscribe("foo")
+        assert p.subscribed is True
+        with pytest.deprecated_call():
+            await p.close()
+        assert p.subscribed is False
+
+    async def test_reset_is_aclose(self, r: redis.Redis):
+        """
+        Test backwards compatible reset method
+        """
+        p = r.pubsub()
+        assert p.subscribed is False
+        await p.subscribe("foo")
+        assert p.subscribed is True
+        with pytest.deprecated_call():
+            await p.reset()
+        assert p.subscribed is False
+
     async def test_ignore_all_subscribe_messages(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
 
@@ -233,7 +272,7 @@ class TestPubSubSubscribeUnsubscribe:
             assert p.subscribed is True
             assert await wait_for_message(p) is None
         assert p.subscribed is False
-        await p.close()
+        await p.aclose()
 
     async def test_ignore_individual_subscribe_messages(self, pubsub):
         p = pubsub
@@ -350,7 +389,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
-        await p.close()
+        await p.aclose()
 
     async def test_channel_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -359,7 +398,7 @@ class TestPubSubMessages:
         assert await r.publish("foo", "test message") == 1
         assert await wait_for_message(p) is None
         assert self.async_message == make_message("message", "foo", "test message")
-        await p.close()
+        await p.aclose()
 
     async def test_channel_sync_async_message_handler(self, r):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -371,7 +410,7 @@ class TestPubSubMessages:
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", "foo", "test message")
         assert self.async_message == make_message("message", "bar", "test message 2")
-        await p.close()
+        await p.aclose()
 
     @pytest.mark.onlynoncluster
     async def test_pattern_message_handler(self, r: redis.Redis):
@@ -383,7 +422,7 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", "foo", "test message", pattern="f*"
         )
-        await p.close()
+        await p.aclose()
 
     async def test_unicode_channel_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -394,7 +433,7 @@ class TestPubSubMessages:
         assert await r.publish(channel, "test message") == 1
         assert await wait_for_message(p) is None
         assert self.message == make_message("message", channel, "test message")
-        await p.close()
+        await p.aclose()
 
     @pytest.mark.onlynoncluster
     # see: https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html
@@ -410,7 +449,7 @@ class TestPubSubMessages:
         assert self.message == make_message(
             "pmessage", channel, "test message", pattern=pattern
         )
-        await p.close()
+        await p.aclose()
 
     async def test_get_message_without_subscribe(self, r: redis.Redis, pubsub):
         p = pubsub
@@ -524,7 +563,7 @@ class TestPubSubAutoDecoding:
         await r.publish(self.channel, new_data)
         assert await wait_for_message(p) is None
         assert self.message == self.make_message("message", self.channel, new_data)
-        await p.close()
+        await p.aclose()
 
     async def test_pattern_message_handler(self, r: redis.Redis):
         p = r.pubsub(ignore_subscribe_messages=True)
@@ -546,7 +585,7 @@ class TestPubSubAutoDecoding:
         assert self.message == self.make_message(
             "pmessage", self.channel, new_data, pattern=self.pattern
         )
-        await p.close()
+        await p.aclose()
 
     async def test_context_manager(self, r: redis.Redis):
         async with r.pubsub() as pubsub:
@@ -556,7 +595,7 @@ class TestPubSubAutoDecoding:
         assert pubsub.connection is None
         assert pubsub.channels == {}
         assert pubsub.patterns == {}
-        await pubsub.close()
+        await pubsub.aclose()
 
 
 @pytest.mark.onlynoncluster
@@ -597,9 +636,9 @@ class TestPubSubSubcommands:
 
         channels = [(b"foo", 1), (b"bar", 2), (b"baz", 3)]
         assert await r.pubsub_numsub("foo", "bar", "baz") == channels
-        await p1.close()
-        await p2.close()
-        await p3.close()
+        await p1.aclose()
+        await p2.aclose()
+        await p3.aclose()
 
     @skip_if_server_version_lt("2.8.0")
     async def test_pubsub_numpat(self, r: redis.Redis):
@@ -608,7 +647,7 @@ class TestPubSubSubcommands:
         for i in range(3):
             assert (await wait_for_message(p))["type"] == "psubscribe"
         assert await r.pubsub_numpat() == 3
-        await p.close()
+        await p.aclose()
 
 
 @pytest.mark.onlynoncluster
@@ -621,7 +660,7 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="", pattern=None
         )
-        await p.close()
+        await p.aclose()
 
     @skip_if_server_version_lt("3.0.0")
     async def test_send_pubsub_ping_message(self, r: redis.Redis):
@@ -631,7 +670,7 @@ class TestPubSubPings:
         assert await wait_for_message(p) == make_message(
             type="pong", channel=None, data="hello world", pattern=None
         )
-        await p.close()
+        await p.aclose()
 
 
 @pytest.mark.onlynoncluster
diff --git a/tests/test_connection_pool.py b/tests/test_connection_pool.py
index ab0fc6be..ef70a8ff 100644
--- a/tests/test_connection_pool.py
+++ b/tests/test_connection_pool.py
@@ -1,6 +1,7 @@
 import os
 import re
 import time
+from contextlib import closing
 from threading import Thread
 from unittest import mock
 
@@ -51,6 +52,16 @@ class TestConnectionPool:
         assert isinstance(connection, DummyConnection)
         assert connection.kwargs == connection_kwargs
 
+    def test_closing(self):
+        connection_kwargs = {"foo": "bar", "biz": "baz"}
+        pool = redis.ConnectionPool(
+            connection_class=DummyConnection,
+            max_connections=None,
+            **connection_kwargs,
+        )
+        with closing(pool):
+            pass
+
     def test_multiple_connections(self, master_host):
         connection_kwargs = {"host": master_host[0], "port": master_host[1]}
         pool = self.get_pool(connection_kwargs=connection_kwargs)
diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py
index 7b048eec..e64a763b 100644
--- a/tests/test_pipeline.py
+++ b/tests/test_pipeline.py
@@ -1,3 +1,6 @@
+from contextlib import closing
+from unittest import mock
+
 import pytest
 import redis
 
@@ -284,6 +287,24 @@ class TestPipeline:
             assert unwatch_command is not None
             assert unwatch_command["command"] == "UNWATCH"
 
+    @pytest.mark.onlynoncluster
+    def test_close_is_reset(self, r):
+        with r.pipeline() as pipe:
+            called = 0
+
+            def mock_reset():
+                nonlocal called
+                called += 1
+
+            with mock.patch.object(pipe, "reset", mock_reset):
+                pipe.close()
+                assert called == 1
+
+    @pytest.mark.onlynoncluster
+    def test_closing(self, r):
+        with closing(r.pipeline()):
+            pass
+
     @pytest.mark.onlynoncluster
     def test_transaction_callable(self, r):
         r["a"] = 1