Merge branch 'master' into bloom-2.4

Author: dvora-h

CHANGES

@@ -1,3 +1,4 @@
+    * Fix timezone handling for datetime to unixtime conversions
     * Fix start_id type for XAUTOCLAIM
     * Remove verbose logging from cluster.py
     * Add retry mechanism to async version of Connection
@@ -16,6 +17,8 @@
     * Added dynaminc_startup_nodes configuration to RedisCluster
     * Fix reusing the old nodes' connections when cluster topology refresh is being done
     * Fix RedisCluster to immediately raise AuthenticationError without a retry
+    * ClusterPipeline Doesn't Handle ConnectionError for Dead Hosts (#2225)
+
 * 4.1.3 (Feb 8, 2022)
     * Fix flushdb and flushall (#1926)
     * Add redis5 and redis4 dockers (#1871)

README.md

@@ -52,7 +52,7 @@ The above code connects to localhost on port 6379, sets a value in Redis, and re
 
 ### Connection Pools
 
-By default, redis-py uses a connection pool to manage connections. Each instance of of a Redis class receives its own connection pool. You can however define your own [redis.ConnectionPool](https://redis.readthedocs.io/en/stable/connections.html#connection-pools)
+By default, redis-py uses a connection pool to manage connections. Each instance of a Redis class receives its own connection pool. You can however define your own [redis.ConnectionPool](https://redis.readthedocs.io/en/stable/connections.html#connection-pools)
 
 ``` python
 >>> pool = redis.ConnectionPool(host='localhost', port=6379, db=0)

redis/cluster.py

@@ -1755,10 +1755,6 @@ def __len__(self):
         """ """
         return len(self.command_stack)
 
-    def __nonzero__(self):
-        "Pipeline instances should  always evaluate to True on Python 2.7"
-        return True
-
     def __bool__(self):
         "Pipeline instances should  always evaluate to True on Python 3+"
         return True
@@ -1901,34 +1897,53 @@ def _send_cluster_commands(
         # we figure out the slot number that command maps to, then from
         # the slot determine the node.
         for c in attempt:
-            # refer to our internal node -> slot table that
-            # tells us where a given
-            # command should route to.
-            passed_targets = c.options.pop("target_nodes", None)
-            if passed_targets and not self._is_nodes_flag(passed_targets):
-                target_nodes = self._parse_target_nodes(passed_targets)
-            else:
-                target_nodes = self._determine_nodes(*c.args, node_flag=passed_targets)
-                if not target_nodes:
+            connection_error_retry_counter = 0
+            while True:
+                # refer to our internal node -> slot table that
+                # tells us where a given command should route to.
+                # (it might be possible we have a cached node that no longer
+                # exists in the cluster, which is why we do this in a loop)
+                passed_targets = c.options.pop("target_nodes", None)
+                if passed_targets and not self._is_nodes_flag(passed_targets):
+                    target_nodes = self._parse_target_nodes(passed_targets)
+                else:
+                    target_nodes = self._determine_nodes(
+                        *c.args, node_flag=passed_targets
+                    )
+                    if not target_nodes:
+                        raise RedisClusterException(
+                            f"No targets were found to execute {c.args} command on"
+                        )
+                if len(target_nodes) > 1:
                     raise RedisClusterException(
-                        f"No targets were found to execute {c.args} command on"
+                        f"Too many targets for command {c.args}"
                     )
-            if len(target_nodes) > 1:
-                raise RedisClusterException(f"Too many targets for command {c.args}")
-
-            node = target_nodes[0]
-            # now that we know the name of the node
-            # ( it's just a string in the form of host:port )
-            # we can build a list of commands for each node.
-            node_name = node.name
-            if node_name not in nodes:
-                redis_node = self.get_redis_connection(node)
-                connection = get_connection(redis_node, c.args)
-                nodes[node_name] = NodeCommands(
-                    redis_node.parse_response, redis_node.connection_pool, connection
-                )
 
-            nodes[node_name].append(c)
+                node = target_nodes[0]
+
+                # now that we know the name of the node
+                # ( it's just a string in the form of host:port )
+                # we can build a list of commands for each node.
+                node_name = node.name
+                if node_name not in nodes:
+                    redis_node = self.get_redis_connection(node)
+                    try:
+                        connection = get_connection(redis_node, c.args)
+                    except ConnectionError:
+                        connection_error_retry_counter += 1
+                        if connection_error_retry_counter < 5:
+                            # reinitialize the node -> slot table
+                            self.nodes_manager.initialize()
+                            continue
+                        else:
+                            raise
+                    nodes[node_name] = NodeCommands(
+                        redis_node.parse_response,
+                        redis_node.connection_pool,
+                        connection,
+                    )
+                nodes[node_name].append(c)
+                break
 
         # send the commands in sequence.
         # we  write to all the open sockets for each node first,

redis/commands/bf/__init__.py

@@ -3,6 +3,7 @@
 from ..helpers import parse_to_list
 from .commands import *  # noqa
 from .info import BFInfo, CFInfo, CMSInfo, TDigestInfo, TopKInfo
+from .utils import parse_tdigest_quantile
 
 
 class AbstractBloom(object):
@@ -166,7 +167,7 @@ def __init__(self, client, **kwargs):
             # TDIGEST_ADD: spaceHolder,
             # TDIGEST_MERGE: spaceHolder,
             TDIGEST_CDF: float,
-            TDIGEST_QUANTILE: float,
+            TDIGEST_QUANTILE: parse_tdigest_quantile,
             TDIGEST_MIN: float,
             TDIGEST_MAX: float,
             TDIGEST_INFO: TDigestInfo,

redis/commands/bf/commands.py

@@ -394,13 +394,14 @@ def max(self, key):
         """  # noqa
         return self.execute_command(TDIGEST_MAX, key)
 
-    def quantile(self, key, quantile):
+    def quantile(self, key, quantile, *quantiles):
         """
-        Return double value estimate of the cutoff such that a specified fraction of the data
-        added to this TDigest would be less than or equal to the cutoff.
+        Returns estimates of one or more cutoffs such that a specified fraction of the
+        observations added to this t-digest would be less than or equal to each of the
+        specified cutoffs. (Multiple quantiles can be returned with one call)
         For more information see `TDIGEST.QUANTILE <https://redis.io/commands/tdigest.quantile>`_.
         """  # noqa
-        return self.execute_command(TDIGEST_QUANTILE, key, quantile)
+        return self.execute_command(TDIGEST_QUANTILE, key, quantile, *quantiles)
 
     def cdf(self, key, value):
         """

redis/commands/bf/utils.py

@@ -0,0 +1,3 @@
+def parse_tdigest_quantile(response):
+    """Parse TDIGEST.QUANTILE response."""
+    return [float(x) for x in response]

redis/commands/core.py

@@ -2,7 +2,6 @@
 
 import datetime
 import hashlib
-import time
 import warnings
 from typing import (
     TYPE_CHECKING,
@@ -17,6 +16,7 @@
     Mapping,
     Optional,
     Sequence,
+    Set,
     Tuple,
     Union,
 )
@@ -1673,7 +1673,7 @@ def expireat(
         For more information see https://redis.io/commands/expireat
         """
         if isinstance(when, datetime.datetime):
-            when = int(time.mktime(when.timetuple()))
+            when = int(when.timestamp())
 
         exp_option = list()
         if nx:
@@ -1768,14 +1768,12 @@ def getex(
         if exat is not None:
             pieces.append("EXAT")
             if isinstance(exat, datetime.datetime):
-                s = int(exat.microsecond / 1000000)
-                exat = int(time.mktime(exat.timetuple())) + s
+                exat = int(exat.timestamp())
             pieces.append(exat)
         if pxat is not None:
             pieces.append("PXAT")
             if isinstance(pxat, datetime.datetime):
-                ms = int(pxat.microsecond / 1000)
-                pxat = int(time.mktime(pxat.timetuple())) * 1000 + ms
+                pxat = int(pxat.timestamp() * 1000)
             pieces.append(pxat)
         if persist:
             pieces.append("PERSIST")
@@ -1994,8 +1992,7 @@ def pexpireat(
         For more information see https://redis.io/commands/pexpireat
         """
         if isinstance(when, datetime.datetime):
-            ms = int(when.microsecond / 1000)
-            when = int(time.mktime(when.timetuple())) * 1000 + ms
+            when = int(when.timestamp() * 1000)
         exp_option = list()
         if nx:
             exp_option.append("NX")
@@ -2196,14 +2193,12 @@ def set(
         if exat is not None:
             pieces.append("EXAT")
             if isinstance(exat, datetime.datetime):
-                s = int(exat.microsecond / 1000000)
-                exat = int(time.mktime(exat.timetuple())) + s
+                exat = int(exat.timestamp())
             pieces.append(exat)
         if pxat is not None:
             pieces.append("PXAT")
             if isinstance(pxat, datetime.datetime):
-                ms = int(pxat.microsecond / 1000)
-                pxat = int(time.mktime(pxat.timetuple())) * 1000 + ms
+                pxat = int(pxat.timestamp() * 1000)
             pieces.append(pxat)
         if keepttl:
             pieces.append("KEEPTTL")
@@ -3257,7 +3252,7 @@ def sismember(self, name: str, value: str) -> Union[Awaitable[bool], bool]:
         """
         return self.execute_command("SISMEMBER", name, value)
 
-    def smembers(self, name: str) -> Union[Awaitable[list], list]:
+    def smembers(self, name: str) -> Union[Awaitable[Set], Set]:
         """
         Return all members of the set ``name``