Code cleanup

Author: Grokzen

rediscluster/__init__.py

@@ -2,7 +2,6 @@
 
 # python std lib
 import logging
-import sys
 
 # rediscluster imports
 from rediscluster.client import RedisCluster

rediscluster/client.py

@@ -2,7 +2,6 @@
 from __future__ import unicode_literals
 
 # python std lib
-import datetime
 import json
 import logging
 import random
@@ -45,11 +44,10 @@
 from redis import Redis
 from redis.client import list_or_args, parse_info
 from redis.connection import Connection, SSLConnection
-from redis._compat import iteritems, basestring, izip, nativestr, long
+from redis._compat import iteritems, nativestr, long
 from redis.exceptions import (
     BusyLoadingError,
     ConnectionError,
-    DataError,
     RedisError,
     ResponseError,
     TimeoutError,
@@ -211,7 +209,7 @@ class RedisCluster(Redis):
         "ZCARD",
         "ZCOUNT",
         "ZRANGE",
-        "ZSCORE"
+        "ZSCORE",
     ]
 
     RESULT_CALLBACKS = dict_merge(
@@ -377,7 +375,7 @@ def __init__(self, host=None, port=None, startup_nodes=None, max_connections=Non
                 nodemanager_follow_cluster=nodemanager_follow_cluster,
                 connection_class=connection_class,
                 host_port_remap=host_port_remap,
-                **kwargs
+                **kwargs,
             )
 
         super(RedisCluster, self).__init__(connection_pool=pool, **kwargs)
@@ -447,7 +445,7 @@ def pubsub(self, **kwargs):
         """
         return ClusterPubSub(self.connection_pool, **kwargs)
 
-    def pipeline(self, transaction=None, shard_hint=None):
+    def pipeline(self, transaction=None, shard_hint=None, read_from_replicas=False):
         """
         Cluster impl:
             Pipelines do not work in cluster mode the same way they do in normal mode.
@@ -466,6 +464,7 @@ def pipeline(self, transaction=None, shard_hint=None):
             result_callbacks=self.result_callbacks,
             response_callbacks=self.response_callbacks,
             cluster_down_retry_attempts=self.cluster_down_retry_attempts,
+            read_from_replicas=read_from_replicas,
         )
 
     def transaction(self, *args, **kwargs):
@@ -492,7 +491,7 @@ def _determine_slot(self, *args):
 
         if command in ['XREADGROUP', 'XREAD']:
             stream_idx = args.index(b'STREAMS')
-            keys_ids = list(args[stream_idx + 1: ])
+            keys_ids = list(args[stream_idx + 1:])
             idx_split = len(keys_ids) // 2
             keys = keys_ids[: idx_split]
             slots = {self.connection_pool.nodes.keyslot(key) for key in keys}
@@ -642,10 +641,10 @@ def _execute_command(self, *args, **kwargs):
                 # This is the last attempt before we run out of TTL, raise the exception
                 if ttl == 1:
                     raise e
-            except (RedisClusterException, BusyLoadingError) as e:
+            except (RedisClusterException, BusyLoadingError):
                 log.exception("RedisClusterException || BusyLoadingError")
                 raise
-            except ConnectionError as e:
+            except ConnectionError:
                 log.exception("ConnectionError")
 
                 connection.disconnect()
@@ -667,8 +666,7 @@ def _execute_command(self, *args, **kwargs):
                     self.connection_pool.nodes.increment_reinitialize_counter(
                         count=self.connection_pool.nodes.reinitialize_steps,
                     )
-
-            except TimeoutError as e:
+            except TimeoutError:
                 log.exception("TimeoutError")
 
                 if ttl < self.RedisClusterRequestTTL / 2:
@@ -695,7 +693,7 @@ def _execute_command(self, *args, **kwargs):
 
                 node = self.connection_pool.nodes.set_node(e.host, e.port, server_type='master')
                 self.connection_pool.nodes.slots[e.slot_id][0] = node
-            except TryAgainError as e:
+            except TryAgainError:
                 log.exception("TryAgainError")
 
                 if ttl < self.RedisClusterRequestTTL / 2:
@@ -733,7 +731,7 @@ def _execute_command_on_nodes(self, nodes, *args, **kwargs):
 
                 connection.send_command(*args)
                 res[node["name"]] = self.parse_response(connection, command, **kwargs)
-            except ClusterDownError as e:
+            except ClusterDownError:
                 self.connection_pool.disconnect()
                 self.connection_pool.reset()
                 self.refresh_table_asap = True

rediscluster/connection.py

@@ -102,7 +102,7 @@ def on_connect(self):
 
             if nativestr(self.read_response()) != 'OK':
                 raise ConnectionError('READONLY command failed')
- 
+
 
 class ClusterConnectionPool(ConnectionPool):
     """
@@ -328,7 +328,7 @@ def get_connection_by_slot(self, slot):
 
         try:
             return self.get_connection_by_node(self.get_node_by_slot(slot))
-        except (KeyError, RedisClusterException) as exc:
+        except (KeyError, RedisClusterException):
             return self.get_random_connection()
 
     def get_connection_by_node(self, node):
@@ -353,7 +353,7 @@ def get_master_node_by_slot(self, slot):
         """
         try:
             return self.nodes.slots[slot][0]
-        except KeyError as ke:
+        except KeyError:
             raise SlotNotCoveredError('Slot "{slot}" not covered by the cluster. "skip_full_coverage_check={skip_full_coverage_check}"'.format(
                 slot=slot, skip_full_coverage_check=self.nodes._skip_full_coverage_check,
             ))
@@ -397,11 +397,11 @@ class ClusterBlockingConnectionPool(ClusterConnectionPool):
         # not available.
         >>> pool = ClusterBlockingConnectionPool(timeout=5)
     """
+
     def __init__(self, startup_nodes=None, init_slot_cache=True, connection_class=None,
                  max_connections=None, max_connections_per_node=False, reinitialize_steps=None,
                  skip_full_coverage_check=False, nodemanager_follow_cluster=False,
                  timeout=20, **connection_kwargs):
-
         self.timeout = timeout
 
         super(ClusterBlockingConnectionPool, self).__init__(
@@ -413,7 +413,7 @@ def __init__(self, startup_nodes=None, init_slot_cache=True, connection_class=No
             reinitialize_steps=reinitialize_steps,
             skip_full_coverage_check=skip_full_coverage_check,
             nodemanager_follow_cluster=nodemanager_follow_cluster,
-            **connection_kwargs
+            **connection_kwargs,
         )
 
     def _blocking_pool_factory(self):
@@ -557,7 +557,8 @@ def __init__(self, startup_nodes=None, init_slot_cache=True, connection_class=No
             max_connections=max_connections,
             readonly=True,
             nodemanager_follow_cluster=nodemanager_follow_cluster,
-            **connection_kwargs)
+            **connection_kwargs,
+        )
 
         self.master_node_commands = ('SCAN', 'SSCAN', 'HSCAN', 'ZSCAN')
 

rediscluster/exceptions.py

@@ -44,6 +44,7 @@ class ClusterCrossSlotError(ResponseError):
 class ClusterDownError(ClusterError, ResponseError):
     """
     """
+
     def __init__(self, resp):
         self.args = (resp, )
         self.message = resp
@@ -72,6 +73,7 @@ def __init__(self, resp):
 class TryAgainError(ResponseError):
     """
     """
+
     def __init__(self, *args, **kwargs):
         pass
 
@@ -81,6 +83,7 @@ class MovedError(AskError):
     """
     pass
 
+
 class MasterDownError(ClusterDownError):
     """
     """

rediscluster/nodemanager.py

@@ -21,7 +21,7 @@
 class NodeManager(object):
     """
     """
-    RedisClusterHashSlots = 16384
+    REDIS_CLUSTER_HASH_SLOTS = 16384
 
     def __init__(self, startup_nodes=None, reinitialize_steps=None, skip_full_coverage_check=False, nodemanager_follow_cluster=False,
                  host_port_remap=None, **connection_kwargs):
@@ -86,7 +86,7 @@ def _validate_host_port_remap(self, host_port_remap):
                 socket.inet_aton(item.get('to_host', '0.0.0.0').strip())
             except socket.error:
                 raise RedisClusterConfigError("Both from_host and to_host in host_port_remap rule must be a valid ip address")
-            if len(item.get('from_host', '0.0.0.0').split('.')) < 4 or len(item.get('to_host', '0.0.0.0').split('.')) < 4 :
+            if len(item.get('from_host', '0.0.0.0').split('.')) < 4 or len(item.get('to_host', '0.0.0.0').split('.')) < 4:
                 raise RedisClusterConfigError(
                     "Both from_host and to_host in host_port_remap rule must must have all octets specified")
 
@@ -110,7 +110,7 @@ def keyslot(self, key):
             if end > -1 and end != start + 1:
                 k = k[start + 1:end]
 
-        return crc16(k) % self.RedisClusterHashSlots
+        return crc16(k) % self.REDIS_CLUSTER_HASH_SLOTS
 
     def node_from_slot(self, slot):
         """
@@ -286,7 +286,7 @@ def initialize(self):
                 need_full_slots_coverage = self.cluster_require_full_coverage(nodes_cache)
 
             # Validate if all slots are covered or if we should try next startup node
-            for i in range(0, self.RedisClusterHashSlots):
+            for i in range(0, self.REDIS_CLUSTER_HASH_SLOTS):
                 if i not in tmp_slots and need_full_slots_coverage:
                     all_slots_covered = False
 
@@ -299,7 +299,7 @@ def initialize(self):
 
         if not all_slots_covered:
             raise RedisClusterException("All slots are not covered after query all startup_nodes. {0} of {1} covered...".format(
-                len(tmp_slots), self.RedisClusterHashSlots))
+                len(tmp_slots), self.REDIS_CLUSTER_HASH_SLOTS))
 
         # Set the tmp variables to the real variables
         self.slots = tmp_slots

rediscluster/pipeline.py

@@ -156,7 +156,7 @@ def send_cluster_commands(self, stack, raise_on_error=True, allow_redirections=T
 
         It will try the number of times specified by the config option "self.cluster_down_retry_attempts"
         which defaults to 3 unless manually configured.
-        
+
         If it reaches the number of times, the command will raises ClusterDownException.
         """
         for _ in range(0, self.cluster_down_retry_attempts):

rediscluster/utils.py

@@ -1,11 +1,8 @@
 # -*- coding: utf-8 -*-
 from socket import gethostbyaddr
-from functools import wraps
 
 # rediscluster imports
-from .exceptions import (
-    RedisClusterException, ClusterDownError
-)
+from .exceptions import RedisClusterException
 
 # 3rd party imports
 from redis._compat import basestring, nativestr

tests/conftest.py

@@ -2,8 +2,8 @@
 
 # python std lib
 import os
+import random
 import sys
-import json
 
 # rediscluster imports
 from rediscluster import RedisCluster
@@ -13,7 +13,6 @@
 from distutils.version import StrictVersion
 from mock import Mock
 from redis import Redis
-from redis.exceptions import ResponseError
 
 # put our path in front so we can be sure we are testing locally not against the global package
 basepath = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
@@ -223,35 +222,38 @@ def mock_cluster_resp_int(request, **kwargs):
 @pytest.fixture()
 def mock_cluster_resp_info(request, **kwargs):
     r = _get_client(RedisCluster, request, **kwargs)
-    response = ('cluster_state:ok\r\ncluster_slots_assigned:16384\r\n'
-                'cluster_slots_ok:16384\r\ncluster_slots_pfail:0\r\n'
-                'cluster_slots_fail:0\r\ncluster_known_nodes:7\r\n'
-                'cluster_size:3\r\ncluster_current_epoch:7\r\n'
-                'cluster_my_epoch:2\r\ncluster_stats_messages_sent:170262\r\n'
-                'cluster_stats_messages_received:105653\r\n')
+    response = (
+        'cluster_state:ok\r\ncluster_slots_assigned:16384\r\n'
+        'cluster_slots_ok:16384\r\ncluster_slots_pfail:0\r\n'
+        'cluster_slots_fail:0\r\ncluster_known_nodes:7\r\n'
+        'cluster_size:3\r\ncluster_current_epoch:7\r\n'
+        'cluster_my_epoch:2\r\ncluster_stats_messages_sent:170262\r\n'
+        'cluster_stats_messages_received:105653\r\n'
+    )
     return _gen_cluster_mock_resp(r, response)
 
 
 @pytest.fixture()
 def mock_cluster_resp_nodes(request, **kwargs):
     r = _get_client(RedisCluster, request, **kwargs)
-    response = ('c8253bae761cb1ecb2b61857d85dfe455a0fec8b 172.17.0.7:7006 '
-                'slave aa90da731f673a99617dfe930306549a09f83a6b 0 '
-                '1447836263059 5 connected\n'
-                '9bd595fe4821a0e8d6b99d70faa660638a7612b3 172.17.0.7:7008 '
-                'master - 0 1447836264065 0 connected\n'
-                'aa90da731f673a99617dfe930306549a09f83a6b 172.17.0.7:7003 '
-                'myself,master - 0 0 2 connected 5461-10922\n'
-                '1df047e5a594f945d82fc140be97a1452bcbf93e 172.17.0.7:7007 '
-                'slave 19efe5a631f3296fdf21a5441680f893e8cc96ec 0 '
-                '1447836262556 3 connected\n'
-                '4ad9a12e63e8f0207025eeba2354bcf4c85e5b22 172.17.0.7:7005 '
-                'master - 0 1447836262555 7 connected 0-5460\n'
-                '19efe5a631f3296fdf21a5441680f893e8cc96ec 172.17.0.7:7004 '
-                'master - 0 1447836263562 3 connected 10923-16383\n'
-                'fbb23ed8cfa23f17eaf27ff7d0c410492a1093d6 172.17.0.7:7002 '
-                'master,fail - 1447829446956 1447829444948 1 disconnected\n'
-                )
+    response = (
+        'c8253bae761cb1ecb2b61857d85dfe455a0fec8b 172.17.0.7:7006 '
+        'slave aa90da731f673a99617dfe930306549a09f83a6b 0 '
+        '1447836263059 5 connected\n'
+        '9bd595fe4821a0e8d6b99d70faa660638a7612b3 172.17.0.7:7008 '
+        'master - 0 1447836264065 0 connected\n'
+        'aa90da731f673a99617dfe930306549a09f83a6b 172.17.0.7:7003 '
+        'myself,master - 0 0 2 connected 5461-10922\n'
+        '1df047e5a594f945d82fc140be97a1452bcbf93e 172.17.0.7:7007 '
+        'slave 19efe5a631f3296fdf21a5441680f893e8cc96ec 0 '
+        '1447836262556 3 connected\n'
+        '4ad9a12e63e8f0207025eeba2354bcf4c85e5b22 172.17.0.7:7005 '
+        'master - 0 1447836262555 7 connected 0-5460\n'
+        '19efe5a631f3296fdf21a5441680f893e8cc96ec 172.17.0.7:7004 '
+        'master - 0 1447836263562 3 connected 10923-16383\n'
+        'fbb23ed8cfa23f17eaf27ff7d0c410492a1093d6 172.17.0.7:7002 '
+        'master,fail - 1447829446956 1447829444948 1 disconnected\n'
+    )
     return _gen_cluster_mock_resp(r, response)