From 2d67a12b2f2ee7fee68b72f5503b3e1a7053ae6c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Fri, 25 Aug 2023 11:23:13 +0000
Subject: [PATCH 1/7] Add the `from_pool` argument to asyncio.Redis

---
 docs/examples/asyncio_examples.ipynb | 33 +++++++++++++++++----
 redis/asyncio/client.py              | 44 ++++++++++++++++++----------
 redis/asyncio/connection.py          |  2 +-
 redis/asyncio/sentinel.py            | 14 +++------
 4 files changed, 62 insertions(+), 31 deletions(-)

diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index f7e67e2ca7..d47b7d6e6e 100644
--- a/docs/examples/asyncio_examples.ipynb
+++ b/docs/examples/asyncio_examples.ipynb
@@ -44,6 +44,26 @@
     "await connection.close()"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "If you create custom `ConnectionPool` for the `Redis` instance to use, use the `from_pool` argument.  This will cause the pool to be disconnected along with the Redis instance. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "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()"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {
@@ -53,7 +73,8 @@
     }
    },
    "source": [
-    "If you supply a custom `ConnectionPool` that is supplied to several `Redis` instances, you may want to disconnect the connection pool explicitly. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
+    "\n",
+    "However, If you supply a `ConnectionPool` that is shared several `Redis` instances, you may want to disconnect the connection pool explicitly. use the `connection_pool` argument in that case."
    ]
   },
   {
@@ -69,10 +90,12 @@
    "source": [
     "import redis.asyncio as redis\n",
     "\n",
-    "connection = redis.Redis(auto_close_connection_pool=False)\n",
-    "await connection.close()\n",
-    "# Or: await connection.close(close_connection_pool=False)\n",
-    "await connection.connection_pool.disconnect()"
+    "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()"
    ]
   },
   {
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index f0c1ab7536..7d6cebb7b2 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -114,7 +114,6 @@ def from_url(
         cls,
         url: str,
         single_connection_client: bool = False,
-        auto_close_connection_pool: bool = True,
         **kwargs,
     ):
         """
@@ -160,12 +159,10 @@ class initializer. In the case of conflicting arguments, querystring
 
         """
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        redis = cls(
-            connection_pool=connection_pool,
+        return cls(
+            from_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
-        redis.auto_close_connection_pool = auto_close_connection_pool
-        return redis
 
     def __init__(
         self,
@@ -179,6 +176,7 @@ def __init__(
         socket_keepalive: Optional[bool] = None,
         socket_keepalive_options: Optional[Mapping[int, Union[int, bytes]]] = None,
         connection_pool: Optional[ConnectionPool] = None,
+        from_pool: Optional[ConnectionPool] = None,
         unix_socket_path: Optional[str] = None,
         encoding: str = "utf-8",
         encoding_errors: str = "strict",
@@ -200,6 +198,7 @@ def __init__(
         lib_version: Optional[str] = get_lib_version(),
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
+        # deprecated. create a pool and use connection_pool instead
         auto_close_connection_pool: bool = True,
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
@@ -213,14 +212,9 @@ 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 not connection_pool:
+
+        if not connection_pool and not from_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,8 +271,28 @@ def __init__(
                             "ssl_check_hostname": ssl_check_hostname,
                         }
                     )
-            connection_pool = ConnectionPool(**kwargs)
-        self.connection_pool = connection_pool
+            # backwards compatibility.  This arg only used if no pool
+            # is provided
+            if auto_close_connection_pool:
+                from_pool = ConnectionPool(**kwargs)
+            else:
+                connection_pool = ConnectionPool(**kwargs)
+
+        if from_pool is not None:
+            # internal connection pool, expected to be closed by Redis instance
+            if connection_pool is not None:
+                raise ValueError(
+                    "Cannot use both from_pool and connection_pool arguments"
+                )
+            self.connection_pool = from_pool
+            self.auto_close_connection_pool = True  # the Redis instance closes the pool
+        else:
+            # external connection pool, expected to be closed by caller
+            self.connection_pool = connection_pool
+            self.auto_close_connection_pool = (
+                False  # the user is expected to close the 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
index 00865515ab..59ffc03962 100644
--- a/redis/asyncio/connection.py
+++ b/redis/asyncio/connection.py
@@ -1107,7 +1107,7 @@ class BlockingConnectionPool(ConnectionPool):
     A blocking connection pool::
 
         >>> from redis.asyncio.client import Redis
-        >>> client = Redis(connection_pool=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
index 20c0a1fdd5..e8d9f8ed2c 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -340,12 +340,9 @@ 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,
+        return redis_class(
+            from_pool=connection_pool,
         )
-        client.auto_close_connection_pool = auto_close_connection_pool
-        return client
 
     def slave_for(
         self,
@@ -377,9 +374,6 @@ 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,
+        return redis_class(
+            from_pool=connection_pool,
         )
-        client.auto_close_connection_pool = auto_close_connection_pool
-        return client

From b99491aa7232a5e16e5c49f6cd110d638fa7825a Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Thu, 24 Aug 2023 15:22:52 +0000
Subject: [PATCH 2/7] Add tests for the `from_pool` argument

---
 tests/test_asyncio/test_connection.py | 73 ++++++++++++++++++++++++---
 1 file changed, 67 insertions(+), 6 deletions(-)

diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 9ee6db1566..5e1fffd031 100644
--- 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,16 +305,13 @@ 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):
+async def test_pool_auto_close_disable(request):
     """Verify that auto_close_connection_pool can be disabled"""
 
     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)
 
@@ -322,3 +319,67 @@ async def 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 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()

From 2758374b6e720917e30eec2549fdb60662da94cc Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Thu, 24 Aug 2023 15:41:58 +0000
Subject: [PATCH 3/7] Add a "from_pool" argument for the sync client too

---
 redis/client.py          | 24 +++++++++--
 tests/test_connection.py | 89 +++++++++++++++++++++++++++++++++++++++-
 2 files changed, 108 insertions(+), 5 deletions(-)

diff --git a/redis/client.py b/redis/client.py
index f695cef534..71c7e7c228 100755
--- a/redis/client.py
+++ b/redis/client.py
@@ -137,7 +137,7 @@ 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(
-            connection_pool=connection_pool,
+            from_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
 
@@ -152,6 +152,7 @@ def __init__(
         socket_keepalive=None,
         socket_keepalive_options=None,
         connection_pool=None,
+        from_pool=None,
         unix_socket_path=None,
         encoding="utf-8",
         encoding_errors="strict",
@@ -198,7 +199,7 @@ def __init__(
             if `True`, connection pool is not used. In that case `Redis`
             instance use is not thread safe.
         """
-        if not connection_pool:
+        if not connection_pool and not from_pool:
             if charset is not None:
                 warnings.warn(
                     DeprecationWarning(
@@ -274,8 +275,20 @@ def __init__(
                             "ssl_ocsp_expected_cert": ssl_ocsp_expected_cert,
                         }
                     )
-            connection_pool = ConnectionPool(**kwargs)
-        self.connection_pool = connection_pool
+            from_pool = ConnectionPool(**kwargs)
+        
+        if from_pool is not None:
+            # internal connection pool, expected to be closed by Redis instance
+            if connection_pool is not None:
+                raise ValueError(
+                    "Cannot use both from_pool and connection_pool arguments")
+            self.connection_pool = from_pool
+            self.auto_close_connection_pool = True  # the Redis instance closes the pool
+        else:
+            # external connection pool, expected to be closed by caller
+            self.connection_pool = connection_pool
+            self.auto_close_connection_pool = False  # the user is expected to close the pool
+
         self.connection = None
         if single_connection_client:
             self.connection = self.connection_pool.get_connection("_")
@@ -477,6 +490,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/tests/test_connection.py b/tests/test_connection.py
index 760b23c9c1..66d83d38fd 100644
--- 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 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()

From d30b77445c74bddd30c35917ad604063b8fd5f12 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Thu, 24 Aug 2023 16:00:11 +0000
Subject: [PATCH 4/7] Add automatic connection pool close for redis.sentinel

---
 redis/sentinel.py      |  8 ++------
 tests/test_sentinel.py | 26 ++++++++++++++++++++++++++
 2 files changed, 28 insertions(+), 6 deletions(-)

diff --git a/redis/sentinel.py b/redis/sentinel.py
index 836e781e7f..312bf75649 100644
--- a/redis/sentinel.py
+++ b/redis/sentinel.py
@@ -354,9 +354,7 @@ def master_for(
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
         return redis_class(
-            connection_pool=connection_pool_class(
-                service_name, self, **connection_kwargs
-            )
+            from_pool=connection_pool_class(service_name, self, **connection_kwargs)
         )
 
     def slave_for(
@@ -387,7 +385,5 @@ def slave_for(
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
         return redis_class(
-            connection_pool=connection_pool_class(
-                service_name, self, **connection_kwargs
-            )
+            from_pool=connection_pool_class(service_name, self, **connection_kwargs)
         )
diff --git a/tests/test_sentinel.py b/tests/test_sentinel.py
index b7bcc27de2..54b9647098 100644
--- 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()

From af81ba8dd338441546a314499c7f6120cc8b63e1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Fri, 25 Aug 2023 09:36:34 +0000
Subject: [PATCH 5/7] use from_pool() class method instead

---
 docs/examples/asyncio_examples.ipynb  |  4 +--
 redis/asyncio/client.py               | 51 ++++++++++++++-------------
 redis/asyncio/connection.py           |  4 +--
 redis/asyncio/sentinel.py             |  8 ++---
 redis/client.py                       | 43 +++++++++++++---------
 redis/sentinel.py                     |  8 ++---
 tests/test_asyncio/test_connection.py | 30 ++++++++++++++--
 tests/test_connection.py              |  4 +--
 8 files changed, 93 insertions(+), 59 deletions(-)

diff --git a/docs/examples/asyncio_examples.ipynb b/docs/examples/asyncio_examples.ipynb
index d47b7d6e6e..a8ee8faf8b 100644
--- a/docs/examples/asyncio_examples.ipynb
+++ b/docs/examples/asyncio_examples.ipynb
@@ -48,7 +48,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "If you create custom `ConnectionPool` for the `Redis` instance to use, use the `from_pool` argument.  This will cause the pool to be disconnected along with the Redis instance. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
+    "If you create custom `ConnectionPool` for the `Redis` instance to use alone, use the `from_pool` class method to create it.  This will cause the pool to be disconnected along with the Redis instance. Disconnecting the connection pool simply disconnects all connections hosted in the pool."
    ]
   },
   {
@@ -60,7 +60,7 @@
     "import redis.asyncio as redis\n",
     "\n",
     "pool = redis.ConnectionPool.from_url(\"redis://localhost\")\n",
-    "connection = redis.Redis(from_pool=pool)\n",
+    "connection = redis.Redis.from_pool(pool)\n",
     "await connection.close()"
    ]
   },
diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index 7d6cebb7b2..c3bbe96563 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -159,10 +159,28 @@ class initializer. In the case of conflicting arguments, querystring
 
         """
         connection_pool = ConnectionPool.from_url(url, **kwargs)
-        return cls(
-            from_pool=connection_pool,
+        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,
@@ -176,7 +194,6 @@ def __init__(
         socket_keepalive: Optional[bool] = None,
         socket_keepalive_options: Optional[Mapping[int, Union[int, bytes]]] = None,
         connection_pool: Optional[ConnectionPool] = None,
-        from_pool: Optional[ConnectionPool] = None,
         unix_socket_path: Optional[str] = None,
         encoding: str = "utf-8",
         encoding_errors: str = "strict",
@@ -213,7 +230,7 @@ def __init__(
         """
         kwargs: Dict[str, Any]
 
-        if not connection_pool and not from_pool:
+        if not connection_pool:
             # Create internal connection pool, expected to be closed by Redis instance
             if not retry_on_error:
                 retry_on_error = []
@@ -271,28 +288,14 @@ def __init__(
                             "ssl_check_hostname": ssl_check_hostname,
                         }
                     )
-            # backwards compatibility.  This arg only used if no pool
-            # is provided
-            if auto_close_connection_pool:
-                from_pool = ConnectionPool(**kwargs)
-            else:
-                connection_pool = ConnectionPool(**kwargs)
-
-        if from_pool is not None:
-            # internal connection pool, expected to be closed by Redis instance
-            if connection_pool is not None:
-                raise ValueError(
-                    "Cannot use both from_pool and connection_pool arguments"
-                )
-            self.connection_pool = from_pool
-            self.auto_close_connection_pool = True  # the Redis instance closes the pool
+            # This arg only used if no pool is passed in
+            self.auto_close_connection_pool = auto_close_connection_pool
+            connection_pool = ConnectionPool(**kwargs)
         else:
-            # external connection pool, expected to be closed by caller
-            self.connection_pool = connection_pool
-            self.auto_close_connection_pool = (
-                False  # the user is expected to close the pool
-            )
+            # 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
index 59ffc03962..4f20e6ec1d 100644
--- 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(from_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
index e8d9f8ed2c..6834fb194f 100644
--- a/redis/asyncio/sentinel.py
+++ b/redis/asyncio/sentinel.py
@@ -340,9 +340,7 @@ def master_for(
 
         connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
         # The Redis object "owns" the pool
-        return redis_class(
-            from_pool=connection_pool,
-        )
+        return redis_class.from_pool(connection_pool)
 
     def slave_for(
         self,
@@ -374,6 +372,4 @@ def slave_for(
 
         connection_pool = connection_pool_class(service_name, self, **connection_kwargs)
         # The Redis object "owns" the pool
-        return redis_class(
-            from_pool=connection_pool,
-        )
+        return redis_class.from_pool(connection_pool)
diff --git a/redis/client.py b/redis/client.py
index 71c7e7c228..017f8ef11d 100755
--- 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(
-            from_pool=connection_pool,
+        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,
@@ -152,7 +170,6 @@ def __init__(
         socket_keepalive=None,
         socket_keepalive_options=None,
         connection_pool=None,
-        from_pool=None,
         unix_socket_path=None,
         encoding="utf-8",
         encoding_errors="strict",
@@ -199,7 +216,7 @@ def __init__(
             if `True`, connection pool is not used. In that case `Redis`
             instance use is not thread safe.
         """
-        if not connection_pool and not from_pool:
+        if not connection_pool:
             if charset is not None:
                 warnings.warn(
                     DeprecationWarning(
@@ -275,20 +292,12 @@ def __init__(
                             "ssl_ocsp_expected_cert": ssl_ocsp_expected_cert,
                         }
                     )
-            from_pool = ConnectionPool(**kwargs)
-        
-        if from_pool is not None:
-            # internal connection pool, expected to be closed by Redis instance
-            if connection_pool is not None:
-                raise ValueError(
-                    "Cannot use both from_pool and connection_pool arguments")
-            self.connection_pool = from_pool
-            self.auto_close_connection_pool = True  # the Redis instance closes the pool
+            connection_pool = ConnectionPool(**kwargs)
+            self.auto_close_connection_pool = True
         else:
-            # external connection pool, expected to be closed by caller
-            self.connection_pool = connection_pool
-            self.auto_close_connection_pool = False  # the user is expected to close the pool
+            self.auto_close_connection_pool = False
 
+        self.connection_pool = connection_pool
         self.connection = None
         if single_connection_client:
             self.connection = self.connection_pool.get_connection("_")
diff --git a/redis/sentinel.py b/redis/sentinel.py
index 312bf75649..41f308d1ee 100644
--- a/redis/sentinel.py
+++ b/redis/sentinel.py
@@ -353,8 +353,8 @@ def master_for(
         kwargs["is_master"] = True
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
-        return redis_class(
-            from_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(
@@ -384,6 +384,6 @@ def slave_for(
         kwargs["is_master"] = False
         connection_kwargs = dict(self.connection_kwargs)
         connection_kwargs.update(kwargs)
-        return redis_class(
-            from_pool=connection_pool_class(service_name, self, **connection_kwargs)
+        return redis_class.from_pool(
+            connection_pool_class(service_name, self, **connection_kwargs)
         )
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index 5e1fffd031..d904e2c1a3 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -355,7 +355,7 @@ async def mock_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 using `from_pool`
+    """Verify that basic Redis instances created using `from_pool()`
     have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--redis-url")
@@ -369,7 +369,7 @@ async def get_redis_connection():
             pool = ConnectionPool.from_url(url)
         else:
             pool = ConnectionPool(**url_args)
-        return Redis(from_pool=pool)
+        return Redis.from_pool(pool)
 
     called = 0
 
@@ -383,3 +383,29 @@ async def mock_disconnect(_):
 
     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():
+        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
index 66d83d38fd..bff249559e 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -268,7 +268,7 @@ def mock_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 using `from_pool`
+    """Verify that basic Redis instances created using `from_pool()`
     have auto_close_connection_pool set to True"""
 
     url: str = request.config.getoption("--redis-url")
@@ -282,7 +282,7 @@ def get_redis_connection():
             pool = ConnectionPool.from_url(url)
         else:
             pool = ConnectionPool(**url_args)
-        return Redis(from_pool=pool)
+        return Redis.from_pool(pool)
 
     called = 0
 

From fb64a6513d0ffb2ddfb6ffe7c21cc3c917e84f07 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Mon, 11 Sep 2023 20:21:29 +0000
Subject: [PATCH 6/7] re-add the auto_close_connection_pool arg to
 Connection.from_url()

---
 redis/asyncio/client.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index c3bbe96563..b28df10cac 100644
--- a/redis/asyncio/client.py
+++ b/redis/asyncio/client.py
@@ -114,6 +114,7 @@ def from_url(
         cls,
         url: str,
         single_connection_client: bool = False,
+        auto_close_connection_pool: bool = True,
         **kwargs,
     ):
         """
@@ -163,7 +164,13 @@ class initializer. In the case of conflicting arguments, querystring
             connection_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
-        client.auto_close_connection_pool = True
+        # We should probably deprecate the `auto_close_connection_pool`
+        # argument.
+        # If the caller doesn't want the pool auto-closed, a better
+        # pattern is to create the pool manually (maybe using from_url()),
+        # pass it in using the `connection_pool`, and hold on to it to close
+        # it later.
+        client.auto_close_connection_pool = auto_close_connection_pool
         return client
 
     @classmethod

From b34c8903222f461575ef5eca25340e696c408d5b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Kristj=C3=A1n=20Valur=20J=C3=B3nsson?= <sweskman@gmail.com>
Date: Sun, 17 Sep 2023 11:18:21 +0000
Subject: [PATCH 7/7] Deprecate the "auto_close_connection_pool" argument to
 Redis() and Redis.from_url()

---
 redis/asyncio/client.py               | 33 ++++++++++++++++++++-------
 tests/test_asyncio/test_connection.py | 15 +++++++++---
 2 files changed, 37 insertions(+), 11 deletions(-)

diff --git a/redis/asyncio/client.py b/redis/asyncio/client.py
index b28df10cac..cce1b5815d 100644
--- 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,
     ):
         """
@@ -164,12 +164,17 @@ class initializer. In the case of conflicting arguments, querystring
             connection_pool=connection_pool,
             single_connection_client=single_connection_client,
         )
-        # We should probably deprecate the `auto_close_connection_pool`
-        # argument.
-        # If the caller doesn't want the pool auto-closed, a better
-        # pattern is to create the pool manually (maybe using from_url()),
-        # pass it in using the `connection_pool`, and hold on to it to close
-        # it later.
+        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
 
@@ -223,7 +228,7 @@ def __init__(
         username: Optional[str] = None,
         retry: Optional[Retry] = None,
         # deprecated. create a pool and use connection_pool instead
-        auto_close_connection_pool: bool = True,
+        auto_close_connection_pool: Optional[bool] = None,
         redis_connect_func=None,
         credential_provider: Optional[CredentialProvider] = None,
         protocol: Optional[int] = 2,
@@ -237,6 +242,18 @@ def __init__(
         """
         kwargs: Dict[str, Any]
 
+        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:
diff --git a/tests/test_asyncio/test_connection.py b/tests/test_asyncio/test_connection.py
index d904e2c1a3..a954a40dbf 100644
--- a/tests/test_asyncio/test_connection.py
+++ b/tests/test_asyncio/test_connection.py
@@ -305,15 +305,23 @@ async def get_redis_connection():
     await r1.close()
 
 
+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"""
+    """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():
         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
@@ -394,7 +402,8 @@ async def test_redis_pool_auto_close_arg(request, auto_close):
     pool = ConnectionPool.from_url(url)
 
     async def get_redis_connection():
-        client = Redis(connection_pool=pool, auto_close_connection_pool=auto_close)
+        with pytest.deprecated_call():
+            client = Redis(connection_pool=pool, auto_close_connection_pool=auto_close)
         return client
 
     called = 0