Merge branch 'master' into redis-pool

Author: dvora-h

CHANGES

@@ -1,3 +1,6 @@
+    * Fix string cleanse in Redis Graph
+    * Make PythonParser resumable in case of error (#2510)
+    * Add `timeout=None` in `SentinelConnectionManager.read_response`
     * Documentation fix: password protected socket connection (#2374)
     * Allow `timeout=None` in `PubSub.get_message()` to wait forever
     * add `nowait` flag to `asyncio.Connection.disconnect()`
@@ -31,6 +34,7 @@
     * Fix Sentinel.execute_command doesn't execute across the entire sentinel cluster bug (#2458)
     * Added a replacement for the default cluster node in the event of failure (#2463)
     * Simplified connection allocation code for asyncio.connection.BlockingConnectionPool
+    * Fix for Unhandled exception related to self.host with unix socket (#2496)
 
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)

README.md

@@ -21,6 +21,12 @@ redis-py 4.3.x will be the last generation of redis-py to support python 3.6 as
 
 ## Installation
 
+Start a redis via docker:
+
+``` bash
+docker run -p 6379:6379 -it redis/redis-stack:latest
+```
+
 To install redis-py, simply:
 
 ``` bash

redis/__init__.py

@@ -53,7 +53,10 @@ def int_or_str(value):
     __version__ = "99.99.99"
 
 
-VERSION = tuple(map(int_or_str, __version__.split(".")))
+try:
+    VERSION = tuple(map(int_or_str, __version__.split(".")))
+except AttributeError:
+    VERSION = tuple(99, 99, 99)
 
 __all__ = [
     "AuthenticationError",

redis/asyncio/client.py

@@ -453,7 +453,7 @@ def __del__(self, _warnings: Any = warnings) -> None:
                 f"Unclosed client session {self!r}", ResourceWarning, source=self
             )
             context = {"client": self, "message": self._DEL_MESSAGE}
-            asyncio.get_event_loop().call_exception_handler(context)
+            asyncio.get_running_loop().call_exception_handler(context)
 
     async def close(self, close_connection_pool: Optional[bool] = None) -> None:
         """
@@ -483,8 +483,8 @@ async def _send_command_parse_response(self, conn, command_name, *args, **option
     async def _disconnect_raise(self, conn: Connection, error: Exception):
         """
         Close the connection and raise an exception
-        if retry_on_timeout is not set or the error
-        is not a TimeoutError
+        if retry_on_error is not set or the error
+        is not one of the specified error types
         """
         await conn.disconnect()
         if (
@@ -798,7 +798,7 @@ async def check_health(self):
 
         if (
             conn.health_check_interval
-            and asyncio.get_event_loop().time() > conn.next_health_check
+            and asyncio.get_running_loop().time() > conn.next_health_check
         ):
             await conn.send_command(
                 "PING", self.HEALTH_CHECK_MESSAGE, check_health=False

redis/asyncio/connection.py

@@ -208,11 +208,18 @@ async def read_response(
 class PythonParser(BaseParser):
     """Plain Python parsing class"""
 
-    __slots__ = BaseParser.__slots__ + ("encoder",)
+    __slots__ = BaseParser.__slots__ + ("encoder", "_buffer", "_pos", "_chunks")
 
     def __init__(self, socket_read_size: int):
         super().__init__(socket_read_size)
         self.encoder: Optional[Encoder] = None
+        self._buffer = b""
+        self._chunks = []
+        self._pos = 0
+
+    def _clear(self):
+        self._buffer = b""
+        self._chunks.clear()
 
     def on_connect(self, connection: "Connection"):
         """Called when the stream connects"""
@@ -227,8 +234,11 @@ def on_disconnect(self):
         if self._stream is not None:
             self._stream = None
         self.encoder = None
+        self._clear()
 
     async def can_read_destructive(self) -> bool:
+        if self._buffer:
+            return True
         if self._stream is None:
             raise RedisError("Buffer is closed.")
         try:
@@ -237,14 +247,23 @@ async def can_read_destructive(self) -> bool:
         except asyncio.TimeoutError:
             return False
 
-    async def read_response(
+    async def read_response(self, disable_decoding: bool = False):
+        if self._chunks:
+            # augment parsing buffer with previously read data
+            self._buffer += b"".join(self._chunks)
+            self._chunks.clear()
+        self._pos = 0
+        response = await self._read_response(disable_decoding=disable_decoding)
+        # Successfully parsing a response allows us to clear our parsing buffer
+        self._clear()
+        return response
+
+    async def _read_response(
         self, disable_decoding: bool = False
     ) -> Union[EncodableT, ResponseError, None]:
         if not self._stream or not self.encoder:
             raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
         raw = await self._readline()
-        if not raw:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
         response: Any
         byte, response = raw[:1], raw[1:]
 
@@ -258,6 +277,7 @@ async def read_response(
             # if the error is a ConnectionError, raise immediately so the user
             # is notified
             if isinstance(error, ConnectionError):
+                self._clear()  # Successful parse
                 raise error
             # otherwise, we're dealing with a ResponseError that might belong
             # inside a pipeline response. the connection's read_response()
@@ -282,7 +302,7 @@ async def read_response(
             if length == -1:
                 return None
             response = [
-                (await self.read_response(disable_decoding)) for _ in range(length)
+                (await self._read_response(disable_decoding)) for _ in range(length)
             ]
         if isinstance(response, bytes) and disable_decoding is False:
             response = self.encoder.decode(response)
@@ -293,25 +313,38 @@ async def _read(self, length: int) -> bytes:
         Read `length` bytes of data.  These are assumed to be followed
         by a '\r\n' terminator which is subsequently discarded.
         """
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-        try:
-            data = await self._stream.readexactly(length + 2)
-        except asyncio.IncompleteReadError as error:
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error
-        return data[:-2]
+        want = length + 2
+        end = self._pos + want
+        if len(self._buffer) >= end:
+            result = self._buffer[self._pos : end - 2]
+        else:
+            tail = self._buffer[self._pos :]
+            try:
+                data = await self._stream.readexactly(want - len(tail))
+            except asyncio.IncompleteReadError as error:
+                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) from error
+            result = (tail + data)[:-2]
+            self._chunks.append(data)
+        self._pos += want
+        return result
 
     async def _readline(self) -> bytes:
         """
         read an unknown number of bytes up to the next '\r\n'
         line separator, which is discarded.
         """
-        if self._stream is None:
-            raise RedisError("Buffer is closed.")
-        data = await self._stream.readline()
-        if not data.endswith(b"\r\n"):
-            raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
-        return data[:-2]
+        found = self._buffer.find(b"\r\n", self._pos)
+        if found >= 0:
+            result = self._buffer[self._pos : found]
+        else:
+            tail = self._buffer[self._pos :]
+            data = await self._stream.readline()
+            if not data.endswith(b"\r\n"):
+                raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
+            result = (tail + data)[:-2]
+            self._chunks.append(data)
+        self._pos += len(result) + 2
+        return result
 
 
 class HiredisParser(BaseParser):
@@ -532,7 +565,7 @@ def repr_pieces(self):
     def __del__(self):
         try:
             if self.is_connected:
-                loop = asyncio.get_event_loop()
+                loop = asyncio.get_running_loop()
                 coro = self.disconnect()
                 if loop.is_running():
                     loop.create_task(coro)

redis/asyncio/lock.py

@@ -201,14 +201,14 @@ async def acquire(
             blocking_timeout = self.blocking_timeout
         stop_trying_at = None
         if blocking_timeout is not None:
-            stop_trying_at = asyncio.get_event_loop().time() + blocking_timeout
+            stop_trying_at = asyncio.get_running_loop().time() + blocking_timeout
         while True:
             if await self.do_acquire(token):
                 self.local.token = token
                 return True
             if not blocking:
                 return False
-            next_try_at = asyncio.get_event_loop().time() + sleep
+            next_try_at = asyncio.get_running_loop().time() + sleep
             if stop_trying_at is not None and next_try_at > stop_trying_at:
                 return False
             await asyncio.sleep(sleep)

redis/asyncio/sentinel.py

@@ -1,7 +1,7 @@
 import asyncio
 import random
 import weakref
-from typing import AsyncIterator, Iterable, Mapping, Sequence, Tuple, Type
+from typing import AsyncIterator, Iterable, Mapping, Optional, Sequence, Tuple, Type
 
 from redis.asyncio.client import Redis
 from redis.asyncio.connection import (
@@ -63,9 +63,16 @@ async def connect(self):
             lambda error: asyncio.sleep(0),
         )
 
-    async def read_response(self, disable_decoding: bool = False):
+    async def read_response(
+        self,
+        disable_decoding: bool = False,
+        timeout: Optional[float] = None,
+    ):
         try:
-            return await super().read_response(disable_decoding=disable_decoding)
+            return await super().read_response(
+                disable_decoding=disable_decoding,
+                timeout=timeout,
+            )
         except ReadOnlyError:
             if self.connection_pool.is_master:
                 # When talking to a master, a ReadOnlyError when likely

redis/cluster.py

@@ -121,6 +121,7 @@ def parse_cluster_shards(resp, **options):
     "charset",
     "connection_class",
     "connection_pool",
+    "connection_pool_class",
     "client_name",
     "credential_provider",
     "db",
@@ -1267,6 +1268,7 @@ def __init__(
         require_full_coverage=False,
         lock=None,
         dynamic_startup_nodes=True,
+        connection_pool_class=ConnectionPool,
         **kwargs,
     ):
         self.nodes_cache = {}
@@ -1277,6 +1279,7 @@ def __init__(
         self.from_url = from_url
         self._require_full_coverage = require_full_coverage
         self._dynamic_startup_nodes = dynamic_startup_nodes
+        self.connection_pool_class = connection_pool_class
         self._moved_exception = None
         self.connection_kwargs = kwargs
         self.read_load_balancer = LoadBalancer()
@@ -1420,7 +1423,7 @@ def create_redis_node(self, host, port, **kwargs):
             # Create a redis node with a costumed connection pool
             kwargs.update({"host": host})
             kwargs.update({"port": port})
-            r = Redis(connection_pool=ConnectionPool(**kwargs))
+            r = Redis(connection_pool=self.connection_pool_class(**kwargs))
         else:
             r = Redis(host=host, port=port, **kwargs)
         return r
@@ -1437,6 +1440,8 @@ def _get_or_create_cluster_node(self, host, port, role, tmp_nodes_cache):
             if target_node is None or target_node.redis_connection is None:
                 # create new cluster node for this cluster
                 target_node = ClusterNode(host, port, role)
+            if target_node.server_type != role:
+                target_node.server_type = role
 
         return target_node
 

redis/commands/bf/__init__.py

@@ -198,6 +198,7 @@ def __init__(self, client, **kwargs):
             # BF_MEXISTS: spaceHolder,
             # BF_SCANDUMP: spaceHolder,
             # BF_LOADCHUNK: spaceHolder,
+            # BF_CARD: spaceHolder,
             BF_INFO: BFInfo,
         }
 

redis/commands/bf/commands.py

@@ -11,6 +11,7 @@
 BF_SCANDUMP = "BF.SCANDUMP"
 BF_LOADCHUNK = "BF.LOADCHUNK"
 BF_INFO = "BF.INFO"
+BF_CARD = "BF.CARD"
 
 CF_RESERVE = "CF.RESERVE"
 CF_ADD = "CF.ADD"
@@ -165,6 +166,14 @@ def info(self, key):
         """  # noqa
         return self.execute_command(BF_INFO, key)
 
+    def card(self, key):
+        """
+        Returns the cardinality of a Bloom filter - number of items that were added to a Bloom filter and detected as unique
+        (items that caused at least one bit to be set in at least one sub-filter).
+        For more information see `BF.CARD <https://redis.io/commands/bf.card>`_.
+        """  # noqa
+        return self.execute_command(BF_CARD, key)
+
 
 class CFCommands:
     """Cuckoo Filter commands."""

redis/commands/cluster.py

@@ -52,6 +52,8 @@
     [
         "BITCOUNT",
         "BITPOS",
+        "EVAL_RO",
+        "EVALSHA_RO",
         "EXISTS",
         "GEODIST",
         "GEOHASH",

redis/commands/core.py

@@ -193,6 +193,7 @@ def acl_setuser(
         selectors: Optional[Iterable[Tuple[str, KeyT]]] = None,
         reset: bool = False,
         reset_keys: bool = False,
+        reset_channels: bool = False,
         reset_passwords: bool = False,
         **kwargs,
     ) -> ResponseT:
@@ -248,6 +249,12 @@ def acl_setuser(
         key permissions will be kept and any new specified key permissions
         will be applied on top.
 
+        ``reset_channels`` is a boolean indicating whether the user's channel
+        permissions should be reset prior to applying any new channel permissions
+        specified in ``channels``.If this is False, the user's existing
+        channel permissions will be kept and any new specified channel permissions
+        will be applied on top.
+
         ``reset_passwords`` is a boolean indicating whether to remove all
         existing passwords and the 'nopass' flag from the user prior to
         applying any new passwords specified in 'passwords' or
@@ -266,6 +273,9 @@ def acl_setuser(
         if reset_keys:
             pieces.append(b"resetkeys")
 
+        if reset_channels:
+            pieces.append(b"resetchannels")
+
         if reset_passwords:
             pieces.append(b"resetpass")
 
@@ -2255,6 +2265,8 @@ def set(
                 pieces.append(int(ex.total_seconds()))
             elif isinstance(ex, int):
                 pieces.append(ex)
+            elif isinstance(ex, str) and ex.isdigit():
+                pieces.append(int(ex))
             else:
                 raise DataError("ex must be datetime.timedelta or int")
         if px is not None:
@@ -5139,7 +5151,7 @@ def eval_ro(
         """
         The read-only variant of the EVAL command
 
-        Execute the read-only Lue ``script`` specifying the ``numkeys`` the script
+        Execute the read-only Lua ``script`` specifying the ``numkeys`` the script
         will touch and the key names and argument values in ``keys_and_args``.
         Returns the result of the script.
 

redis/commands/graph/edge.py

@@ -61,6 +61,10 @@ def __str__(self):
         return res
 
     def __eq__(self, rhs):
+        # Type checking
+        if not isinstance(rhs, Edge):
+            return False
+
         # Quick positive check, if both IDs are set.
         if self.id is not None and rhs.id is not None and self.id == rhs.id:
             return True