Repository: redis/redis-py (Python)

Issue:
`auto_close_connection_pool` ignored
This defect pertains to issues discovered with #2831 

In async mode, it is important that a `Redis` object which holds its own pool, can be told to close it when the `Redis` object
is closed, to free the user from such hassles.  Examples in the code include
``` python
from redis.asyncio.client import Redis
client = Redis(connection_pool=BlockingConnectionPool())
```
when this client is closed, the internal connection pool remains open, i.e. its connections are not shut down.

The `Redis` class includes a `auto_close_connection_pool` argument.  However, this is _ignored_ when a connection pool
is provided to the constructor.  Presumably the thinking is that the caller will manage that pool himself.
But that makes the above kind of pattern, also used within the code, unusable.

```python
client = Redis(connection_pool=BlockingConnectionPool(), auto_close_connection_pool=True)
assert client.auto_close_connection_pool=False
```

Instead, it is necessary to have code like this, which is very poor practice:
```python
client = Redis(connection_pool=BlockingConnectionPool(), auto_close_connection_pool=True)
client.auto_close_connection_pool=True
```

I propose, that we either:

- Always honour the `auto_close_connection_pool` argument.  
- add a `own_connection_pool:ConnectionPool` argument, to be used instead of `connection_pool`.
- rename `auto_close_connection_pool` to `own_pool`, or something less cumbersome.
- add a `construct_connection_pool` argument which takes a callable which returns a connection pool, one which Redis considers its own.

At any rate, it should be easy to create a connection pool, and hand it to the `Redis` constructor for it to "own" without
having to fudge an internal property afterwards.

In addition, I propose that the same mechanics be added to the non-`async` parts of the library.  Currently the `redis.Redis` objects do not close connection pools, instead the library relies on the prompt execution of `__del__()` methods, or explicit `ConnectionPool.disconnect()` calls.  
Relying on garbage collection to free resources is considered bad programming practice in Python, where resource management should in general be more explicit.  It is particularly frowned upon for async code, since executing asynchronous code there is not a good idea(*).  the synchronous code should receive whatever solution we choose for the above as well, so that pools owned by a `Redis` client object are also disconnected when those Redis objects are closed.

> (*)  `__del__()` methods are not async.  While it is possible to use something like `asyncio.run` within such a function, it will __block__ the rest of the program.  Also, `__del__()` methods are often invoked during program cleanup when it is simply not possible to do anything and trying to do so will produce errors and confusion.

### Update

In pr #2913 I have updated a `Redis.from_pool()` method to use when handing over a `ConnectionPool` to be _owned_ by the `Redis` instance.  I think this is a much nicer way than to heap on more constructor arguments to `Redis`


Hints:

Patch:
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -114,7 +114,7 @@ def from_url(
         cls,
         url: str,
         single_connection_client: bool = False,
-        auto_close_connection_pool: bool = True,
+        auto_close_connection_pool: Optional[bool] = None,
         **kwargs,
     ):
         """
@@ -160,12 +160,39 @@ class initializer. In the case of conflicting arguments, querystring
 
         """
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        redis = cls(
+        client = cls(
             connection_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
-        redis.auto_close_connection_pool = auto_close_connection_pool
-        return redis
+        if auto_close_connection_pool is not None:
+            warnings.warn(
+                DeprecationWarning(
+                    '"auto_close_connection_pool" is deprecated '
+                    "since version 5.0.0. "
+                    "Please create a ConnectionPool explicitly and "
+                    "provide to the Redis() constructor instead."
+                )
+            )
+        else:
+            auto_close_connection_pool = True
+        client.auto_close_connection_pool = auto_close_connection_pool
+        return client
+
+    @classmethod
+    def from_pool(
+        cls: Type["Redis"],
+        connection_pool: ConnectionPool,
+    ) -> "Redis":
+        """
+        Return a Redis client from the given connection pool.
+        The Redis client will take ownership of the connection pool and
+        close it when the Redis client is closed.
+        """
+        client = cls(
+            connection_pool=connection_pool,
+        )
+        client.auto_close_connection_pool = True
+        return client
 
     def __init__(
         self,
@@ -200,7 +227,8 @@ def __init__(
         lib_version: Optional[str] = get_lib_version(),
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
-        auto_close_connection_pool: bool = True,
+        # deprecated. create a pool and use connection_pool instead
+        auto_close_connection_pool: Optional[bool] = None,
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
         protocol: Optional[int] = 2,
@@ -213,14 +241,21 @@ def __init__(
         To retry on TimeoutError, `retry_on_timeout` can also be set to `True`.
         """
         kwargs: Dict[str, Any]
-        # auto_close_connection_pool only has an effect if connection_pool is
-        # None. This is a similar feature to the missing __del__ to resolve #1103,
-        # but it accounts for whether a user wants to manually close the connection
-        # pool, as a similar feature to ConnectionPool's __del__.
-        self.auto_close_connection_pool = (
-            auto_close_connection_pool if connection_pool is None else False
-        )
+
+        if auto_close_connection_pool is not None:
+            warnings.warn(
+                DeprecationWarning(
+                    '"auto_close_connection_pool" is deprecated '
+                    "since version 5.0.0. "
+                    "Please create a ConnectionPool explicitly and "
+                    "provide to the Redis() constructor instead."
+                )
+            )
+        else:
+            auto_close_connection_pool = True
+
         if not connection_pool:
+            # Create internal connection pool, expected to be closed by Redis instance
             if not retry_on_error:
                 retry_on_error = []
             if retry_on_timeout is True:
@@ -277,7 +312,13 @@ def __init__(
                             "ssl_check_hostname": ssl_check_hostname,
                         }
                     )
+            # This arg only used if no pool is passed in
+            self.auto_close_connection_pool = auto_close_connection_pool
             connection_pool = ConnectionPool(**kwargs)
+        else:
+            # If a pool is passed in, do not close it
+            self.auto_close_connection_pool = False
+
         self.connection_pool = connection_pool
         self.single_connection_client = single_connection_client
         self.connection: Optional[Connection] = None
diff --git a/redis/asyncio/connection.py b/redis/asyncio/connection.py
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1106,8 +1106,8 @@ class BlockingConnectionPool(ConnectionPool):
     """
     A blocking connection pool::
 
-        >>> from redis.asyncio.client import Redis
-        >>> client = Redis(connection_pool=BlockingConnectionPool())
+        >>> from redis.asyncio import Redis, BlockingConnectionPool
+        >>> client = Redis.from_pool(BlockingConnectionPool())
 
     It performs the same function as the default
     :py:class:`~redis.asyncio.ConnectionPool` implementation, in that,
diff --git a/redis/asyncio/sentinel.py b/redis/asyncio/sentinel.py
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -340,12 +340,7 @@ def master_for(
 
         connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
         # The Redis object "owns" the pool
-        auto_close_connection_pool = True
-        client = redis_class(
-            connection_pool=connection_pool,
-        )
-        client.auto_close_connection_pool = auto_close_connection_pool
-        return client
+        return redis_class.from_pool(connection_pool)
 
     def slave_for(
         self,
@@ -377,9 +372,4 @@ def slave_for(
 
         connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
         # The Redis object "owns" the pool
-        auto_close_connection_pool = True
-        client = redis_class(
-            connection_pool=connection_pool,
-        )
-        client.auto_close_connection_pool = auto_close_connection_pool
-        return client
+        return redis_class.from_pool(connection_pool)
diff --git a/redis/client.py b/redis/client.py
--- a/redis/client.py
+++ b/redis/client.py
@@ -4,7 +4,7 @@
 import time
 import warnings
 from itertools import chain
-from typing import Optional
+from typing import Optional, Type
 
 from redis._parsers.helpers import (
     _RedisCallbacks,
@@ -136,10 +136,28 @@ class initializer. In the case of conflicting arguments, querystring
         """
         single_connection_client = kwargs.pop("single_connection_client", False)
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        return cls(
+        client = cls(
             connection_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
+        client.auto_close_connection_pool = True
+        return client
+
+    @classmethod
+    def from_pool(
+        cls: Type["Redis"],
+        connection_pool: ConnectionPool,
+    ) -> "Redis":
+        """
+        Return a Redis client from the given connection pool.
+        The Redis client will take ownership of the connection pool and
+        close it when the Redis client is closed.
+        """
+        client = cls(
+            connection_pool=connection_pool,
+        )
+        client.auto_close_connection_pool = True
+        return client
 
     def __init__(
         self,
@@ -275,6 +293,10 @@ def __init__(
                         }
                     )
             connection_pool = ConnectionPool(**kwargs)
+            self.auto_close_connection_pool = True
+        else:
+            self.auto_close_connection_pool = False
+
         self.connection_pool = connection_pool
         self.connection = None
         if single_connection_client:
@@ -477,6 +499,9 @@ def close(self):
             self.connection = None
             self.connection_pool.release(conn)
 
+        if self.auto_close_connection_pool:
+            self.connection_pool.disconnect()
+
     def _send_command_parse_response(self, conn, command_name, *args, **options):
         """
         Send a command and parse the response
diff --git a/redis/sentinel.py b/redis/sentinel.py
--- a/redis/sentinel.py
+++ b/redis/sentinel.py
@@ -353,10 +353,8 @@ def master_for(
         kwargs["is_master"] = True
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
-        return redis_class(
-            connection_pool=connection_pool_class(
-                service_name, self, **connection_kwargs
-            )
+        return redis_class.from_pool(
+            connection_pool_class(service_name, self, **connection_kwargs)
         )
 
     def slave_for(
@@ -386,8 +384,6 @@ def slave_for(
         kwargs["is_master"] = False
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
-        return redis_class(
-            connection_pool=connection_pool_class(
-                service_name, self, **connection_kwargs
-            )
+        return redis_class.from_pool(
+            connection_pool_class(service_name, self, **connection_kwargs)
         )


Test patch:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -11,7 +11,7 @@
     _AsyncRESP3Parser,
     _AsyncRESPBase,
 )
-from redis.asyncio import Redis
+from redis.asyncio import ConnectionPool, Redis
 from redis.asyncio.connection import Connection, UnixDomainSocketConnection, parse_url
 from redis.asyncio.retry import Retry
 from redis.backoff import NoBackoff
@@ -288,7 +288,7 @@ def test_create_single_connection_client_from_url():
     assert client.single_connection_client is True
 
 
-@pytest.mark.parametrize("from_url", (True, False))
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
 async def test_pool_auto_close(request, from_url):
     """Verify that basic Redis instances have auto_close_connection_pool set to True"""
 
@@ -305,20 +305,116 @@ async def get_redis_connection():
     await r1.close()
 
 
-@pytest.mark.parametrize("from_url", (True, False))
-async def test_pool_auto_close_disable(request, from_url):
-    """Verify that auto_close_connection_pool can be disabled"""
+async def test_pool_from_url_deprecation(request):
+    url: str = request.config.getoption("--redis-url")
+
+    with pytest.deprecated_call():
+        return Redis.from_url(url, auto_close_connection_pool=False)
+
+
+async def test_pool_auto_close_disable(request):
+    """Verify that auto_close_connection_pool can be disabled (deprecated)"""
 
     url: str = request.config.getoption("--redis-url")
     url_args = parse_url(url)
 
     async def get_redis_connection():
-        if from_url:
-            return Redis.from_url(url, auto_close_connection_pool=False)
         url_args["auto_close_connection_pool"] = False
-        return Redis(**url_args)
+        with pytest.deprecated_call():
+            return Redis(**url_args)
 
     r1 = await get_redis_connection()
     assert r1.auto_close_connection_pool is False
     await r1.connection_pool.disconnect()
     await r1.close()
+
+
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
+async def test_redis_connection_pool(request, from_url):
+    """Verify that basic Redis instances using `connection_pool`
+    have auto_close_connection_pool set to False"""
+
+    url: str = request.config.getoption("--redis-url")
+    url_args = parse_url(url)
+
+    pool = None
+
+    async def get_redis_connection():
+        nonlocal pool
+        if from_url:
+            pool = ConnectionPool.from_url(url)
+        else:
+            pool = ConnectionPool(**url_args)
+        return Redis(connection_pool=pool)
+
+    called = 0
+
+    async def mock_disconnect(_):
+        nonlocal called
+        called += 1
+
+    with patch.object(ConnectionPool, "disconnect", mock_disconnect):
+        async with await get_redis_connection() as r1:
+            assert r1.auto_close_connection_pool is False
+
+    assert called == 0
+    await pool.disconnect()
+
+
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
+async def test_redis_from_pool(request, from_url):
+    """Verify that basic Redis instances created using `from_pool()`
+    have auto_close_connection_pool set to True"""
+
+    url: str = request.config.getoption("--redis-url")
+    url_args = parse_url(url)
+
+    pool = None
+
+    async def get_redis_connection():
+        nonlocal pool
+        if from_url:
+            pool = ConnectionPool.from_url(url)
+        else:
+            pool = ConnectionPool(**url_args)
+        return Redis.from_pool(pool)
+
+    called = 0
+
+    async def mock_disconnect(_):
+        nonlocal called
+        called += 1
+
+    with patch.object(ConnectionPool, "disconnect", mock_disconnect):
+        async with await get_redis_connection() as r1:
+            assert r1.auto_close_connection_pool is True
+
+    assert called == 1
+    await pool.disconnect()
+
+
+@pytest.mark.parametrize("auto_close", (True, False))
+async def test_redis_pool_auto_close_arg(request, auto_close):
+    """test that redis instance where pool is provided have
+    auto_close_connection_pool set to False, regardless of arg"""
+
+    url: str = request.config.getoption("--redis-url")
+    pool = ConnectionPool.from_url(url)
+
+    async def get_redis_connection():
+        with pytest.deprecated_call():
+            client = Redis(connection_pool=pool, auto_close_connection_pool=auto_close)
+        return client
+
+    called = 0
+
+    async def mock_disconnect(_):
+        nonlocal called
+        called += 1
+
+    with patch.object(ConnectionPool, "disconnect", mock_disconnect):
+        async with await get_redis_connection() as r1:
+            assert r1.auto_close_connection_pool is False
+
+    assert called == 0
+    await pool.disconnect()
diff --git a/tests/test_connection.py b/tests/test_connection.py
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -5,9 +5,15 @@
 
 import pytest
 import redis
+from redis import ConnectionPool, Redis
 from redis._parsers import _HiredisParser, _RESP2Parser, _RESP3Parser
 from redis.backoff import NoBackoff
-from redis.connection import Connection, SSLConnection, UnixDomainSocketConnection
+from redis.connection import (
+    Connection,
+    SSLConnection,
+    UnixDomainSocketConnection,
+    parse_url,
+)
 from redis.exceptions import ConnectionError, InvalidResponse, TimeoutError
 from redis.retry import Retry
 from redis.utils import HIREDIS_AVAILABLE
@@ -209,3 +215,84 @@ def test_create_single_connection_client_from_url():
         "redis://localhost:6379/0?", single_connection_client=True
     )
     assert client.connection is not None
+
+
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
+def test_pool_auto_close(request, from_url):
+    """Verify that basic Redis instances have auto_close_connection_pool set to True"""
+
+    url: str = request.config.getoption("--redis-url")
+    url_args = parse_url(url)
+
+    def get_redis_connection():
+        if from_url:
+            return Redis.from_url(url)
+        return Redis(**url_args)
+
+    r1 = get_redis_connection()
+    assert r1.auto_close_connection_pool is True
+    r1.close()
+
+
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
+def test_redis_connection_pool(request, from_url):
+    """Verify that basic Redis instances using `connection_pool`
+    have auto_close_connection_pool set to False"""
+
+    url: str = request.config.getoption("--redis-url")
+    url_args = parse_url(url)
+
+    pool = None
+
+    def get_redis_connection():
+        nonlocal pool
+        if from_url:
+            pool = ConnectionPool.from_url(url)
+        else:
+            pool = ConnectionPool(**url_args)
+        return Redis(connection_pool=pool)
+
+    called = 0
+
+    def mock_disconnect(_):
+        nonlocal called
+        called += 1
+
+    with patch.object(ConnectionPool, "disconnect", mock_disconnect):
+        with get_redis_connection() as r1:
+            assert r1.auto_close_connection_pool is False
+
+    assert called == 0
+    pool.disconnect()
+
+
+@pytest.mark.parametrize("from_url", (True, False), ids=("from_url", "from_args"))
+def test_redis_from_pool(request, from_url):
+    """Verify that basic Redis instances created using `from_pool()`
+    have auto_close_connection_pool set to True"""
+
+    url: str = request.config.getoption("--redis-url")
+    url_args = parse_url(url)
+
+    pool = None
+
+    def get_redis_connection():
+        nonlocal pool
+        if from_url:
+            pool = ConnectionPool.from_url(url)
+        else:
+            pool = ConnectionPool(**url_args)
+        return Redis.from_pool(pool)
+
+    called = 0
+
+    def mock_disconnect(_):
+        nonlocal called
+        called += 1
+
+    with patch.object(ConnectionPool, "disconnect", mock_disconnect):
+        with get_redis_connection() as r1:
+            assert r1.auto_close_connection_pool is True
+
+    assert called == 1
+    pool.disconnect()
diff --git a/tests/test_sentinel.py b/tests/test_sentinel.py
--- a/tests/test_sentinel.py
+++ b/tests/test_sentinel.py
@@ -1,4 +1,5 @@
 import socket
+from unittest import mock
 
 import pytest
 import redis.sentinel
@@ -240,3 +241,28 @@ def test_flushconfig(cluster, sentinel):
 def test_reset(cluster, sentinel):
     cluster.master["is_odown"] = True
     assert sentinel.sentinel_reset("mymaster")
+
+
+@pytest.mark.onlynoncluster
+@pytest.mark.parametrize("method_name", ["master_for", "slave_for"])
+def test_auto_close_pool(cluster, sentinel, method_name):
+    """
+    Check that the connection pool created by the sentinel client is
+    automatically closed
+    """
+
+    method = getattr(sentinel, method_name)
+    client = method("mymaster", db=9)
+    pool = client.connection_pool
+    assert client.auto_close_connection_pool is True
+    calls = 0
+
+    def mock_disconnect():
+        nonlocal calls
+        calls += 1
+
+    with mock.patch.object(pool, "disconnect", mock_disconnect):
+        client.close()
+
+    assert calls == 1
+    pool.disconnect()