Merge branch 'master' into ckpy312

Author: chayim

.github/CODEOWNERS

@@ -0,0 +1 @@
+doctests/* @dmaier-redislabs

README.md

@@ -17,6 +17,20 @@ The Python interface to the Redis key-value store.
 
 ---------------------------------------------
 
+## How do I Redis?
+
+[Learn for free at Redis University](https://university.redis.com/)
+
+[Build faster with the Redis Launchpad](https://launchpad.redis.com/)
+
+[Try the Redis Cloud](https://redis.com/try-free/)
+
+[Dive in developer tutorials](https://developer.redis.com/)
+
+[Join the Redis community](https://redis.com/community/)
+
+[Work at Redis](https://redis.com/company/careers/jobs/)
+
 ## Installation
 
 Start a redis via docker:

redis/asyncio/client.py

@@ -546,6 +546,7 @@ def __del__(
                 _grl().call_exception_handler(context)
             except RuntimeError:
                 pass
+            self.connection._close()
 
     async def aclose(self, close_connection_pool: Optional[bool] = None) -> None:
         """

redis/asyncio/connection.py

@@ -5,6 +5,7 @@
 import socket
 import ssl
 import sys
+import warnings
 import weakref
 from abc import abstractmethod
 from itertools import chain
@@ -204,6 +205,24 @@ def __init__(
                 raise ConnectionError("protocol must be either 2 or 3")
             self.protocol = protocol
 
+    def __del__(self, _warnings: Any = warnings):
+        # For some reason, the individual streams don't get properly garbage
+        # collected and therefore produce no resource warnings.  We add one
+        # here, in the same style as those from the stdlib.
+        if getattr(self, "_writer", None):
+            _warnings.warn(
+                f"unclosed Connection {self!r}", ResourceWarning, source=self
+            )
+            self._close()
+
+    def _close(self):
+        """
+        Internal method to silently close the connection without waiting
+        """
+        if self._writer:
+            self._writer.close()
+            self._writer = self._reader = None
+
     def __repr__(self):
         repr_args = ",".join((f"{k}={v}" for k, v in self.repr_pieces()))
         return f"{self.__class__.__name__}<{repr_args}>"
@@ -1017,7 +1036,7 @@ def __repr__(self):
 
     def reset(self):
         self._available_connections = []
-        self._in_use_connections = set()
+        self._in_use_connections = weakref.WeakSet()
 
     def can_get_connection(self) -> bool:
         """Return True if a connection can be retrieved from the pool."""
@@ -1027,21 +1046,25 @@ def can_get_connection(self) -> bool:
         )
 
     async def get_connection(self, command_name, *keys, **options):
-        """Get a connection from the pool"""
+        """Get a connected connection from the pool"""
+        connection = self.get_available_connection()
+        try:
+            await self.ensure_connection(connection)
+        except BaseException:
+            await self.release(connection)
+            raise
+
+        return connection
+
+    def get_available_connection(self):
+        """Get a connection from the pool, without making sure it is connected"""
         try:
             connection = self._available_connections.pop()
         except IndexError:
             if len(self._in_use_connections) >= self.max_connections:
                 raise ConnectionError("Too many connections") from None
             connection = self.make_connection()
         self._in_use_connections.add(connection)
-
-        try:
-            await self.ensure_connection(connection)
-        except BaseException:
-            await self.release(connection)
-            raise
-
         return connection
 
     def get_encoder(self):
@@ -1166,13 +1189,21 @@ def __init__(
     async def get_connection(self, command_name, *keys, **options):
         """Gets a connection from the pool, blocking until one is available"""
         try:
-            async with async_timeout(self.timeout):
-                async with self._condition:
+            async with self._condition:
+                async with async_timeout(self.timeout):
                     await self._condition.wait_for(self.can_get_connection)
-                    return await super().get_connection(command_name, *keys, **options)
+                    connection = super().get_available_connection()
         except asyncio.TimeoutError as err:
             raise ConnectionError("No connection available.") from err
 
+        # We now perform the connection check outside of the lock.
+        try:
+            await self.ensure_connection(connection)
+            return connection
+        except BaseException:
+            await self.release(connection)
+            raise
+
     async def release(self, connection: AbstractConnection):
         """Releases the connection back to the pool."""
         async with self._condition:

redis/client.py

@@ -94,7 +94,7 @@ class Redis(RedisModuleCommands, CoreCommands, SentinelCommands):
     """
 
     @classmethod
-    def from_url(cls, url: str, **kwargs) -> None:
+    def from_url(cls, url: str, **kwargs) -> "Redis":
         """
         Return a Redis client object configured from the given URL
 
@@ -1100,7 +1100,7 @@ def handle_message(self, response, ignore_subscribe_messages=False):
 
     def run_in_thread(
         self,
-        sleep_time: int = 0,
+        sleep_time: float = 0.0,
         daemon: bool = False,
         exception_handler: Optional[Callable] = None,
     ) -> "PubSubWorkerThread":

redis/commands/search/field.py

@@ -13,6 +13,7 @@ class Field:
     SORTABLE = "SORTABLE"
     NOINDEX = "NOINDEX"
     AS = "AS"
+    GEOSHAPE = "GEOSHAPE"
 
     def __init__(
         self,
@@ -91,6 +92,21 @@ def __init__(self, name: str, **kwargs):
         Field.__init__(self, name, args=[Field.NUMERIC], **kwargs)
 
 
+class GeoShapeField(Field):
+    """
+    GeoShapeField is used to enable within/contain indexing/searching
+    """
+
+    SPHERICAL = "SPHERICAL"
+    FLAT = "FLAT"
+
+    def __init__(self, name: str, coord_system=None, **kwargs):
+        args = [Field.GEOSHAPE]
+        if coord_system:
+            args.append(coord_system)
+        Field.__init__(self, name, args=args, **kwargs)
+
+
 class GeoField(Field):
     """
     GeoField is used to define a geo-indexing field in a schema definition

tests/test_asyncio/test_commands.py

@@ -370,10 +370,12 @@ async def test_client_setinfo(self, r: redis.Redis):
         info = await r2.client_info()
         assert info["lib-name"] == "test2"
         assert info["lib-ver"] == "1234"
+        await r2.aclose()
         r3 = redis.asyncio.Redis(lib_name=None, lib_version=None)
         info = await r3.client_info()
         assert info["lib-name"] == ""
         assert info["lib-ver"] == ""
+        await r3.aclose()
 
     @skip_if_server_version_lt("2.6.9")
     @pytest.mark.onlynoncluster

tests/test_asyncio/test_connect.py

@@ -73,6 +73,8 @@ async def _handler(reader, writer):
         try:
             return await _redis_request_handler(reader, writer, stop_event)
         finally:
+            writer.close()
+            await writer.wait_closed()
             finished.set()
 
     if isinstance(server_address, str):

tests/test_asyncio/test_connection.py

@@ -85,6 +85,8 @@ async def get_conn(_):
 
     assert init_call_count == 1
     assert command_call_count == 2
+    r.connection = None  # it was a Mock
+    await r.aclose()
 
 
 @skip_if_server_version_lt("4.0.0")
@@ -143,6 +145,7 @@ async def mock_connect():
     conn._connect.side_effect = mock_connect
     await conn.connect()
     assert conn._connect.call_count == 3
+    await conn.disconnect()
 
 
 async def test_connect_without_retry_on_os_error():
@@ -194,6 +197,7 @@ async def test_connection_parse_response_resume(r: redis.Redis):
         pytest.fail("didn't receive a response")
     assert response
     assert i > 0
+    await conn.disconnect()
 
 
 @pytest.mark.onlynoncluster
@@ -316,7 +320,8 @@ async def mock_aclose(self):
     url: str = request.config.getoption("--redis-url")
     r1 = await Redis.from_url(url)
     with patch.object(r1, "aclose", mock_aclose):
-        await r1.close()
+        with pytest.deprecated_call():
+            await r1.close()
         assert calls == 1
 
     with pytest.deprecated_call():
@@ -436,3 +441,52 @@ async def mock_disconnect(_):
 
     assert called == 0
     await pool.disconnect()
+
+
+async def test_client_garbage_collection(request):
+    """
+    Test that a Redis client will call _close() on any
+    connection that it holds at time of destruction
+    """
+
+    url: str = request.config.getoption("--redis-url")
+    pool = ConnectionPool.from_url(url)
+
+    # create a client with a connection from the pool
+    client = Redis(connection_pool=pool, single_connection_client=True)
+    await client.initialize()
+    with mock.patch.object(client, "connection") as a:
+        # we cannot, in unittests, or from asyncio, reliably trigger garbage collection
+        # so we must just invoke the handler
+        with pytest.warns(ResourceWarning):
+            client.__del__()
+            assert a._close.called
+
+    await client.aclose()
+    await pool.aclose()
+
+
+async def test_connection_garbage_collection(request):
+    """
+    Test that a Connection object will call close() on the
+    stream that it holds.
+    """
+
+    url: str = request.config.getoption("--redis-url")
+    pool = ConnectionPool.from_url(url)
+
+    # create a client with a connection from the pool
+    client = Redis(connection_pool=pool, single_connection_client=True)
+    await client.initialize()
+    conn = client.connection
+
+    with mock.patch.object(conn, "_reader"):
+        with mock.patch.object(conn, "_writer") as a:
+            # we cannot, in unittests, or from asyncio, reliably trigger
+            # garbage collection so we must just invoke the handler
+            with pytest.warns(ResourceWarning):
+                conn.__del__()
+                assert a.close.called
+
+    await client.aclose()
+    await pool.aclose()

tests/test_asyncio/test_cwe_404.py

@@ -15,6 +15,8 @@ def __init__(self, addr, redis_addr, delay: float = 0.0):
         self.send_event = asyncio.Event()
         self.server = None
         self.task = None
+        self.cond = asyncio.Condition()
+        self.running = 0
 
     async def __aenter__(self):
         await self.start()
@@ -63,10 +65,10 @@ async def stop(self):
         except asyncio.CancelledError:
             pass
         await self.server.wait_closed()
-        # do we need to close individual connections too?
-        # prudently close all async generators
-        loop = self.server.get_loop()
-        await loop.shutdown_asyncgens()
+        # Server does not wait for all spawned tasks.  We must do that also to ensure
+        # that all sockets are closed.
+        async with self.cond:
+            await self.cond.wait_for(lambda: self.running == 0)
 
     async def pipe(
         self,
@@ -75,6 +77,7 @@ async def pipe(
         name="",
         event: asyncio.Event = None,
     ):
+        self.running += 1
         try:
             while True:
                 data = await reader.read(1000)
@@ -94,6 +97,10 @@ async def pipe(
                 # ignore errors on close pertaining to no event loop. Don't want
                 # to clutter the test output with errors if being garbage collected
                 pass
+            async with self.cond:
+                self.running -= 1
+                if self.running == 0:
+                    self.cond.notify_all()
 
 
 @pytest.mark.onlynoncluster