from redis.crc import key_slot
from redis.exceptions import RedisClusterException, RedisError
from .core import ACLCommands, DataAccessCommands, ManagementCommands, PubSubCommands
from .helpers import list_or_args
class ClusterMultiKeyCommands:
"""
A class containing commands that handle more than one key
"""
def _partition_keys_by_slot(self, keys):
"""
Split keys into a dictionary that maps a slot to
a list of keys.
"""
slots_to_keys = {}
for key in keys:
k = self.encoder.encode(key)
slot = key_slot(k)
slots_to_keys.setdefault(slot, []).append(key)
return slots_to_keys
def mget_nonatomic(self, keys, *args):
"""
Splits the keys into different slots and then calls MGET
for the keys of every slot. This operation will not be atomic
if keys belong to more than one slot.
Returns a list of values ordered identically to ``keys``
"""
from redis.client import EMPTY_RESPONSE
options = {}
if not args:
options[EMPTY_RESPONSE] = []
# Concatenate all keys into a list
keys = list_or_args(keys, args)
# Split keys into slots
slots_to_keys = self._partition_keys_by_slot(keys)
# Call MGET for every slot and concatenate
# the results
# We must make sure that the keys are returned in order
all_results = {}
for slot_keys in slots_to_keys.values():
slot_values = self.execute_command("MGET", *slot_keys, **options)
slot_results = dict(zip(slot_keys, slot_values))
all_results.update(slot_results)
# Sort the results
vals_in_order = [all_results[key] for key in keys]
return vals_in_order
def mset_nonatomic(self, mapping):
"""
Sets key/values based on a mapping. Mapping is a dictionary of
key/value pairs. Both keys and values should be strings or types that
can be cast to a string via str().
Splits the keys into different slots and then calls MSET
for the keys of every slot. This operation will not be atomic
if keys belong to more than one slot.
"""
# Partition the keys by slot
slots_to_pairs = {}
for pair in mapping.items():
# encode the key
k = self.encoder.encode(pair[0])
slot = key_slot(k)
slots_to_pairs.setdefault(slot, []).extend(pair)
# Call MSET for every slot and concatenate
# the results (one result per slot)
res = []
for pairs in slots_to_pairs.values():
res.append(self.execute_command("MSET", *pairs))
return res
def _split_command_across_slots(self, command, *keys):
"""
Runs the given command once for the keys
of each slot. Returns the sum of the return values.
"""
# Partition the keys by slot
slots_to_keys = self._partition_keys_by_slot(keys)
# Sum up the reply from each command
total = 0
for slot_keys in slots_to_keys.values():
total += self.execute_command(command, *slot_keys)
return total
def exists(self, *keys):
"""
Returns the number of ``names`` that exist in the
whole cluster. The keys are first split up into slots
and then an EXISTS command is sent for every slot
"""
return self._split_command_across_slots("EXISTS", *keys)
def delete(self, *keys):
"""
Deletes the given keys in the cluster.
The keys are first split up into slots
and then an DEL command is sent for every slot
Non-existant keys are ignored.
Returns the number of keys that were deleted.
"""
return self._split_command_across_slots("DEL", *keys)
def touch(self, *keys):
"""
Updates the last access time of given keys across the
cluster.
The keys are first split up into slots
and then an TOUCH command is sent for every slot
Non-existant keys are ignored.
Returns the number of keys that were touched.
"""
return self._split_command_across_slots("TOUCH", *keys)
def unlink(self, *keys):
"""
Remove the specified keys in a different thread.
The keys are first split up into slots
and then an TOUCH command is sent for every slot
Non-existant keys are ignored.
Returns the number of keys that were unlinked.
"""
return self._split_command_across_slots("UNLINK", *keys)
class ClusterManagementCommands(ManagementCommands):
"""
A class for Redis Cluster management commands
The class inherits from Redis's core ManagementCommands class and do the
required adjustments to work with cluster mode
"""
def slaveof(self, *args, **kwargs):
raise RedisClusterException("SLAVEOF is not supported in cluster mode")
def replicaof(self, *args, **kwargs):
raise RedisClusterException("REPLICAOF is not supported in cluster" " mode")
def swapdb(self, *args, **kwargs):
raise RedisClusterException("SWAPDB is not supported in cluster" " mode")
class ClusterDataAccessCommands(DataAccessCommands):
"""
A class for Redis Cluster Data Access Commands
The class inherits from Redis's core DataAccessCommand class and do the
required adjustments to work with cluster mode
"""
def stralgo(
self,
algo,
value1,
value2,
specific_argument="strings",
len=False,
idx=False,
minmatchlen=None,
withmatchlen=False,
**kwargs,
):
target_nodes = kwargs.pop("target_nodes", None)
if specific_argument == "strings" and target_nodes is None:
target_nodes = "default-node"
kwargs.update({"target_nodes": target_nodes})
return super().stralgo(
algo,
value1,
value2,
specific_argument,
len,
idx,
minmatchlen,
withmatchlen,
**kwargs,
)
[docs]class RedisClusterCommands(
ClusterMultiKeyCommands,
ClusterManagementCommands,
ACLCommands,
PubSubCommands,
ClusterDataAccessCommands,
):
"""
A class for all Redis Cluster commands
For key-based commands, the target node(s) will be internally determined
by the keys' hash slot.
Non-key-based commands can be executed with the 'target_nodes' argument to
target specific nodes. By default, if target_nodes is not specified, the
command will be executed on the default cluster node.
:param :target_nodes: type can be one of the followings:
- nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
- 'ClusterNode'
- 'list(ClusterNodes)'
- 'dict(any:clusterNodes)'
for example:
r.cluster_info(target_nodes=RedisCluster.ALL_NODES)
"""
[docs] def cluster_addslots(self, target_node, *slots):
"""
Assign new hash slots to receiving node. Sends to specified node.
:target_node: 'ClusterNode'
The node to execute the command on
"""
return self.execute_command(
"CLUSTER ADDSLOTS", *slots, target_nodes=target_node
)
[docs] def cluster_countkeysinslot(self, slot_id):
"""
Return the number of local keys in the specified hash slot
Send to node based on specified slot_id
"""
return self.execute_command("CLUSTER COUNTKEYSINSLOT", slot_id)
[docs] def cluster_count_failure_report(self, node_id):
"""
Return the number of failure reports active for a given node
Sends to a random node
"""
return self.execute_command("CLUSTER COUNT-FAILURE-REPORTS", node_id)
[docs] def cluster_delslots(self, *slots):
"""
Set hash slots as unbound in the cluster.
It determines by it self what node the slot is in and sends it there
Returns a list of the results for each processed slot.
"""
return [self.execute_command("CLUSTER DELSLOTS", slot) for slot in slots]
[docs] def cluster_failover(self, target_node, option=None):
"""
Forces a slave to perform a manual failover of its master
Sends to specified node
:target_node: 'ClusterNode'
The node to execute the command on
"""
if option:
if option.upper() not in ["FORCE", "TAKEOVER"]:
raise RedisError(
f"Invalid option for CLUSTER FAILOVER command: {option}"
)
else:
return self.execute_command(
"CLUSTER FAILOVER", option, target_nodes=target_node
)
else:
return self.execute_command("CLUSTER FAILOVER", target_nodes=target_node)
[docs] def cluster_info(self, target_nodes=None):
"""
Provides info about Redis Cluster node state.
The command will be sent to a random node in the cluster if no target
node is specified.
"""
return self.execute_command("CLUSTER INFO", target_nodes=target_nodes)
[docs] def cluster_keyslot(self, key):
"""
Returns the hash slot of the specified key
Sends to random node in the cluster
"""
return self.execute_command("CLUSTER KEYSLOT", key)
[docs] def cluster_meet(self, host, port, target_nodes=None):
"""
Force a node cluster to handshake with another node.
Sends to specified node.
"""
return self.execute_command(
"CLUSTER MEET", host, port, target_nodes=target_nodes
)
[docs] def cluster_nodes(self):
"""
Force a node cluster to handshake with another node
Sends to random node in the cluster
"""
return self.execute_command("CLUSTER NODES")
[docs] def cluster_replicate(self, target_nodes, node_id):
"""
Reconfigure a node as a slave of the specified master node
"""
return self.execute_command(
"CLUSTER REPLICATE", node_id, target_nodes=target_nodes
)
[docs] def cluster_reset(self, soft=True, target_nodes=None):
"""
Reset a Redis Cluster node
If 'soft' is True then it will send 'SOFT' argument
If 'soft' is False then it will send 'HARD' argument
"""
return self.execute_command(
"CLUSTER RESET", b"SOFT" if soft else b"HARD", target_nodes=target_nodes
)
[docs] def cluster_save_config(self, target_nodes=None):
"""
Forces the node to save cluster state on disk
"""
return self.execute_command("CLUSTER SAVECONFIG", target_nodes=target_nodes)
[docs] def cluster_get_keys_in_slot(self, slot, num_keys):
"""
Returns the number of keys in the specified cluster slot
"""
return self.execute_command("CLUSTER GETKEYSINSLOT", slot, num_keys)
[docs] def cluster_set_config_epoch(self, epoch, target_nodes=None):
"""
Set the configuration epoch in a new node
"""
return self.execute_command(
"CLUSTER SET-CONFIG-EPOCH", epoch, target_nodes=target_nodes
)
[docs] def cluster_setslot(self, target_node, node_id, slot_id, state):
"""
Bind an hash slot to a specific node
:target_node: 'ClusterNode'
The node to execute the command on
"""
if state.upper() in ("IMPORTING", "NODE", "MIGRATING"):
return self.execute_command(
"CLUSTER SETSLOT", slot_id, state, node_id, target_nodes=target_node
)
elif state.upper() == "STABLE":
raise RedisError('For "stable" state please use ' "cluster_setslot_stable")
else:
raise RedisError(f"Invalid slot state: {state}")
[docs] def cluster_setslot_stable(self, slot_id):
"""
Clears migrating / importing state from the slot.
It determines by it self what node the slot is in and sends it there.
"""
return self.execute_command("CLUSTER SETSLOT", slot_id, "STABLE")
[docs] def cluster_replicas(self, node_id, target_nodes=None):
"""
Provides a list of replica nodes replicating from the specified primary
target node.
"""
return self.execute_command(
"CLUSTER REPLICAS", node_id, target_nodes=target_nodes
)
[docs] def cluster_slots(self, target_nodes=None):
"""
Get array of Cluster slot to node mappings
"""
return self.execute_command("CLUSTER SLOTS", target_nodes=target_nodes)
[docs] def readonly(self, target_nodes=None):
"""
Enables read queries.
The command will be sent to the default cluster node if target_nodes is
not specified.
"""
if target_nodes == "replicas" or target_nodes == "all":
# read_from_replicas will only be enabled if the READONLY command
# is sent to all replicas
self.read_from_replicas = True
return self.execute_command("READONLY", target_nodes=target_nodes)
[docs] def readwrite(self, target_nodes=None):
"""
Disables read queries.
The command will be sent to the default cluster node if target_nodes is
not specified.
"""
# Reset read from replicas flag
self.read_from_replicas = False
return self.execute_command("READWRITE", target_nodes=target_nodes)