Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import asyncio
4import warnings
5from typing import (
6 TYPE_CHECKING,
7 Any,
8 AsyncIterator,
9 Awaitable,
10 Dict,
11 Iterable,
12 Iterator,
13 List,
14 Literal,
15 Mapping,
16 NoReturn,
17 Sequence,
18 overload,
19)
21from redis.crc import key_slot
22from redis.exceptions import RedisClusterException, RedisError
23from redis.typing import (
24 AnyKeyT,
25 AsyncClientProtocol,
26 ClusterCommandsProtocol,
27 ClusterLinksResponse,
28 ClusterNodeDetail,
29 ClusterShardsResponse,
30 EncodableT,
31 KeysT,
32 KeyT,
33 PatternT,
34 ResponseT,
35 StralgoResponse,
36 SyncClientProtocol,
37)
38from redis.utils import deprecated_function
40from .core import (
41 ACLCommands,
42 AsyncACLCommands,
43 AsyncBlessCommands,
44 AsyncDataAccessCommands,
45 AsyncFunctionCommands,
46 AsyncManagementCommands,
47 AsyncModuleCommands,
48 AsyncScriptCommands,
49 BlessCommands,
50 BlessFlag,
51 DataAccessCommands,
52 FunctionCommands,
53 HotkeysMetricsTypes,
54 ManagementCommands,
55 ModuleCommands,
56 PubSubCommands,
57 ScriptCommands,
58)
59from .helpers import list_or_args
60from .redismodules import AsyncRedisModuleCommands, RedisModuleCommands
62if TYPE_CHECKING:
63 from redis.asyncio.cluster import TargetNodesT
64 from redis.cluster import LoadBalancingStrategy
66# DEPRECATED - no longer consulted by the default metadata routing, and it will be removed in a future release.
67#
68# Replica safety is now decided from command metadata by the metadata resolver,
69# so this set no longer describes what commands are safe to execute on replicas.
70# It is kept as a public attribute only so an external caller reading it keeps working;
71# editing it changes nothing.
72#
73# To change replica safety, edit ``redis.commands.metadata._STATIC_COMMAND_METADATA`` or pass
74# a ``metadata_resolver`` to the client. Nothing here.
75READ_COMMANDS = frozenset(
76 [
77 # Bit Operations
78 "BITCOUNT",
79 "BITFIELD_RO",
80 "BITPOS",
81 # Scripting
82 "EVAL_RO",
83 "EVALSHA_RO",
84 "FCALL_RO",
85 # Key Operations
86 "DBSIZE",
87 "DIGEST",
88 "DUMP",
89 "EXISTS",
90 "EXPIRETIME",
91 "PEXPIRETIME",
92 "KEYS",
93 "SCAN",
94 "PTTL",
95 "RANDOMKEY",
96 "TTL",
97 "TYPE",
98 # String Operations
99 "GET",
100 "GETBIT",
101 "GETRANGE",
102 "MGET",
103 "STRLEN",
104 "LCS",
105 # Geo Operations
106 "GEODIST",
107 "GEOHASH",
108 "GEOPOS",
109 "GEOSEARCH",
110 # Hash Operations
111 "HEXISTS",
112 "HGET",
113 "HGETALL",
114 "HKEYS",
115 "HLEN",
116 "HMGET",
117 "HSTRLEN",
118 "HVALS",
119 "HRANDFIELD",
120 "HEXPIRETIME",
121 "HPEXPIRETIME",
122 "HTTL",
123 "HPTTL",
124 "HSCAN",
125 # List Operations
126 "LINDEX",
127 "LPOS",
128 "LLEN",
129 "LRANGE",
130 # Set Operations
131 "SCARD",
132 "SDIFF",
133 "SDIFFCARD",
134 "SINTER",
135 "SINTERCARD",
136 "SISMEMBER",
137 "SMISMEMBER",
138 "SMEMBERS",
139 "SRANDMEMBER",
140 "SUNION",
141 "SUNIONCARD",
142 "SSCAN",
143 # Sorted Set Operations
144 "ZCARD",
145 "ZCOUNT",
146 "ZDIFF",
147 "ZINTER",
148 "ZINTERCARD",
149 "ZLEXCOUNT",
150 "ZMSCORE",
151 "ZRANDMEMBER",
152 "ZRANGE",
153 "ZRANGEBYLEX",
154 "ZRANGEBYSCORE",
155 "ZRANK",
156 "ZREVRANGE",
157 "ZREVRANGEBYLEX",
158 "ZREVRANGEBYSCORE",
159 "ZREVRANK",
160 "ZSCAN",
161 "ZSCORE",
162 "ZUNION",
163 # Stream Operations
164 "XLEN",
165 "XPENDING",
166 "XRANGE",
167 "XREAD",
168 "XREVRANGE",
169 # JSON Module
170 "JSON.ARRINDEX",
171 "JSON.ARRLEN",
172 "JSON.GET",
173 "JSON.MGET",
174 "JSON.OBJKEYS",
175 "JSON.OBJLEN",
176 "JSON.RESP",
177 "JSON.STRLEN",
178 "JSON.TYPE",
179 # RediSearch Module
180 "FT.ALIASLIST",
181 "FT.EXPLAIN",
182 "FT.INFO",
183 "FT.PROFILE",
184 "FT.SEARCH",
185 ]
186)
189class ClusterMultiKeyCommands(ClusterCommandsProtocol):
190 """
191 A class containing commands that handle more than one key
192 """
194 # Read-routing configuration, which the cluster clients all set on the instance. The
195 # defaults are here for a host class that mixes these commands in without one - see
196 # ``_is_replica_safe`` below, which is the same fallback for the same audience - so
197 # that ``_execute_pipeline_by_slot`` reads a value rather than raising
198 # ``AttributeError``. Typed under TYPE_CHECKING because ``redis.cluster`` imports this
199 # module.
200 read_from_replicas: bool = False
201 load_balancing_strategy: LoadBalancingStrategy | None = None
203 def _partition_keys_by_slot(self, keys: Iterable[KeyT]) -> Dict[int, List[KeyT]]:
204 """Split keys into a dictionary that maps a slot to a list of keys."""
206 slots_to_keys = {}
207 for key in keys:
208 slot = key_slot(self.encoder.encode(key))
209 slots_to_keys.setdefault(slot, []).append(key)
211 return slots_to_keys
213 def _partition_pairs_by_slot(
214 self, mapping: Mapping[AnyKeyT, EncodableT]
215 ) -> Dict[int, List[EncodableT]]:
216 """Split pairs into a dictionary that maps a slot to a list of pairs."""
218 slots_to_pairs = {}
219 for pair in mapping.items():
220 slot = key_slot(self.encoder.encode(pair[0]))
221 slots_to_pairs.setdefault(slot, []).extend(pair)
223 return slots_to_pairs
225 def _is_replica_safe(self, command_name: str) -> bool:
226 warnings.warn(
227 "Using READ_COMMANDS for replica-safe checks is deprecated and will be removed in a future release. "
228 "Use a MetadataResolver instead.",
229 DeprecationWarning,
230 stacklevel=2,
231 )
232 return isinstance(command_name, str) and command_name.upper() in READ_COMMANDS
234 def _execute_pipeline_by_slot(
235 self, command: str, slots_to_args: Mapping[int, Iterable[EncodableT]]
236 ) -> List[Any]:
237 replica_safe = (
238 self.read_from_replicas or self.load_balancing_strategy is not None
239 ) and self._is_replica_safe(command)
240 pipe = self.pipeline()
241 [
242 pipe.execute_command(
243 command,
244 *slot_args,
245 target_nodes=[
246 self.nodes_manager.get_node_from_slot(
247 slot,
248 replica_safe,
249 self.load_balancing_strategy if replica_safe else None,
250 )
251 ],
252 )
253 for slot, slot_args in slots_to_args.items()
254 ]
255 return pipe.execute()
257 def _reorder_keys_by_command(
258 self,
259 keys: Iterable[KeyT],
260 slots_to_args: Mapping[int, Iterable[EncodableT]],
261 responses: Iterable[Any],
262 ) -> List[Any]:
263 results = {
264 k: v
265 for slot_values, response in zip(slots_to_args.values(), responses)
266 for k, v in zip(slot_values, response)
267 }
268 return [results[key] for key in keys]
270 def mget_nonatomic(self, keys: KeysT, *args: KeyT) -> List[Any | None]:
271 """
272 Splits the keys into different slots and then calls MGET
273 for the keys of every slot. This operation will not be atomic
274 if keys belong to more than one slot.
276 Returns a list of values ordered identically to ``keys``
278 For more information see https://redis.io/commands/mget
279 """
281 # Concatenate all keys into a list
282 keys = list_or_args(keys, args)
284 # Split keys into slots
285 slots_to_keys = self._partition_keys_by_slot(keys)
287 # Execute commands using a pipeline
288 res = self._execute_pipeline_by_slot("MGET", slots_to_keys)
290 # Reorder keys in the order the user provided & return
291 return self._reorder_keys_by_command(keys, slots_to_keys, res)
293 def mset_nonatomic(self, mapping: Mapping[AnyKeyT, EncodableT]) -> List[bool]:
294 """
295 Sets key/values based on a mapping. Mapping is a dictionary of
296 key/value pairs. Both keys and values should be strings or types that
297 can be cast to a string via str().
299 Splits the keys into different slots and then calls MSET
300 for the keys of every slot. This operation will not be atomic
301 if keys belong to more than one slot.
303 For more information see https://redis.io/commands/mset
304 """
306 # Partition the keys by slot
307 slots_to_pairs = self._partition_pairs_by_slot(mapping)
309 # Execute commands using a pipeline & return list of replies
310 return self._execute_pipeline_by_slot("MSET", slots_to_pairs)
312 def _split_command_across_slots(self, command: str, *keys: KeyT) -> int:
313 """
314 Runs the given command once for the keys
315 of each slot. Returns the sum of the return values.
316 """
318 # Partition the keys by slot
319 slots_to_keys = self._partition_keys_by_slot(keys)
321 # Sum up the reply from each command
322 return sum(self._execute_pipeline_by_slot(command, slots_to_keys))
324 @overload
325 def exists(self: SyncClientProtocol, *keys: KeyT) -> int: ...
327 @overload
328 def exists(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
330 def exists(self, *keys: KeyT) -> int | Awaitable[int]:
331 """
332 Returns the number of ``names`` that exist in the
333 whole cluster. The keys are first split up into slots
334 and then an EXISTS command is sent for every slot
336 For more information see https://redis.io/commands/exists
337 """
338 return self._split_command_across_slots("EXISTS", *keys)
340 @overload
341 def delete(self: SyncClientProtocol, *keys: KeyT) -> int: ...
343 @overload
344 def delete(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
346 def delete(self, *keys: KeyT) -> int | Awaitable[int]:
347 """
348 Deletes the given keys in the cluster.
349 The keys are first split up into slots
350 and then an DEL command is sent for every slot
352 Non-existent keys are ignored.
353 Returns the number of keys that were deleted.
355 For more information see https://redis.io/commands/del
356 """
357 return self._split_command_across_slots("DEL", *keys)
359 @overload
360 def touch(self: SyncClientProtocol, *keys: KeyT) -> int: ...
362 @overload
363 def touch(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
365 def touch(self, *keys: KeyT) -> int | Awaitable[int]:
366 """
367 Updates the last access time of given keys across the
368 cluster.
370 The keys are first split up into slots
371 and then an TOUCH command is sent for every slot
373 Non-existent keys are ignored.
374 Returns the number of keys that were touched.
376 For more information see https://redis.io/commands/touch
377 """
378 return self._split_command_across_slots("TOUCH", *keys)
380 @overload
381 def unlink(self: SyncClientProtocol, *keys: KeyT) -> int: ...
383 @overload
384 def unlink(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
386 def unlink(self, *keys: KeyT) -> int | Awaitable[int]:
387 """
388 Remove the specified keys in a different thread.
390 The keys are first split up into slots
391 and then an TOUCH command is sent for every slot
393 Non-existent keys are ignored.
394 Returns the number of keys that were unlinked.
396 For more information see https://redis.io/commands/unlink
397 """
398 return self._split_command_across_slots("UNLINK", *keys)
401class AsyncClusterMultiKeyCommands(ClusterMultiKeyCommands):
402 """
403 A class containing commands that handle more than one key
404 """
406 async def mget_nonatomic(self, keys: KeysT, *args: KeyT) -> List[Any | None]:
407 """
408 Splits the keys into different slots and then calls MGET
409 for the keys of every slot. This operation will not be atomic
410 if keys belong to more than one slot.
412 Returns a list of values ordered identically to ``keys``
414 For more information see https://redis.io/commands/mget
415 """
417 # Concatenate all keys into a list
418 keys = list_or_args(keys, args)
420 # Split keys into slots
421 slots_to_keys = self._partition_keys_by_slot(keys)
423 # Execute commands using a pipeline
424 res = await self._execute_pipeline_by_slot("MGET", slots_to_keys)
426 # Reorder keys in the order the user provided & return
427 return self._reorder_keys_by_command(keys, slots_to_keys, res)
429 async def mset_nonatomic(self, mapping: Mapping[AnyKeyT, EncodableT]) -> List[bool]:
430 """
431 Sets key/values based on a mapping. Mapping is a dictionary of
432 key/value pairs. Both keys and values should be strings or types that
433 can be cast to a string via str().
435 Splits the keys into different slots and then calls MSET
436 for the keys of every slot. This operation will not be atomic
437 if keys belong to more than one slot.
439 For more information see https://redis.io/commands/mset
440 """
442 # Partition the keys by slot
443 slots_to_pairs = self._partition_pairs_by_slot(mapping)
445 # Execute commands using a pipeline & return list of replies
446 return await self._execute_pipeline_by_slot("MSET", slots_to_pairs)
448 async def _split_command_across_slots(self, command: str, *keys: KeyT) -> int:
449 """
450 Runs the given command once for the keys
451 of each slot. Returns the sum of the return values.
452 """
454 # Partition the keys by slot
455 slots_to_keys = self._partition_keys_by_slot(keys)
457 # Sum up the reply from each command
458 return sum(await self._execute_pipeline_by_slot(command, slots_to_keys))
460 async def _is_replica_safe(self, command_name: str) -> bool:
461 warnings.warn(
462 "Using READ_COMMANDS for replica-safe checks is deprecated and will be removed in a future release. "
463 "Use an AsyncMetadataResolver instead.",
464 DeprecationWarning,
465 stacklevel=2,
466 )
467 return isinstance(command_name, str) and command_name.upper() in READ_COMMANDS
469 async def _execute_pipeline_by_slot(
470 self, command: str, slots_to_args: Mapping[int, Iterable[EncodableT]]
471 ) -> List[Any]:
472 if self._initialize:
473 await self.initialize()
474 replica_safe = (
475 self.read_from_replicas or self.load_balancing_strategy is not None
476 ) and await self._is_replica_safe(command)
477 pipe = self.pipeline()
478 [
479 pipe.execute_command(
480 command,
481 *slot_args,
482 target_nodes=[
483 self.nodes_manager.get_node_from_slot(
484 slot,
485 replica_safe,
486 self.load_balancing_strategy if replica_safe else None,
487 )
488 ],
489 )
490 for slot, slot_args in slots_to_args.items()
491 ]
492 return await pipe.execute()
495class ClusterManagementCommands(ManagementCommands):
496 """
497 A class for Redis Cluster management commands
499 The class inherits from Redis's core ManagementCommands class and do the
500 required adjustments to work with cluster mode
501 """
503 def slaveof(self, *args, **kwargs) -> NoReturn:
504 """
505 Make the server a replica of another instance, or promote it as master.
507 For more information see https://redis.io/commands/slaveof
508 """
509 raise RedisClusterException("SLAVEOF is not supported in cluster mode")
511 def replicaof(self, *args, **kwargs) -> NoReturn:
512 """
513 Make the server a replica of another instance, or promote it as master.
515 For more information see https://redis.io/commands/replicaof
516 """
517 raise RedisClusterException("REPLICAOF is not supported in cluster mode")
519 def swapdb(self, *args, **kwargs) -> NoReturn:
520 """
521 Swaps two Redis databases.
523 For more information see https://redis.io/commands/swapdb
524 """
525 raise RedisClusterException("SWAPDB is not supported in cluster mode")
527 @overload
528 def cluster_myid(
529 self: SyncClientProtocol, target_node: "TargetNodesT"
530 ) -> bytes | str: ...
532 @overload
533 def cluster_myid(
534 self: AsyncClientProtocol, target_node: "TargetNodesT"
535 ) -> Awaitable[bytes | str]: ...
537 def cluster_myid(self, target_node: "TargetNodesT") -> (bytes | str) | Awaitable[
538 bytes | str
539 ]:
540 """
541 Returns the node's id.
543 :target_node: 'ClusterNode'
544 The node to execute the command on
546 For more information check https://redis.io/commands/cluster-myid/
547 """
548 return self.execute_command("CLUSTER MYID", target_nodes=target_node)
550 @overload
551 def cluster_addslots(
552 self: SyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
553 ) -> bool: ...
555 @overload
556 def cluster_addslots(
557 self: AsyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
558 ) -> Awaitable[bool]: ...
560 def cluster_addslots(
561 self, target_node: "TargetNodesT", *slots: EncodableT
562 ) -> bool | Awaitable[bool]:
563 """
564 Assign new hash slots to receiving node. Sends to specified node.
566 :target_node: 'ClusterNode'
567 The node to execute the command on
569 For more information see https://redis.io/commands/cluster-addslots
570 """
571 return self.execute_command(
572 "CLUSTER ADDSLOTS", *slots, target_nodes=target_node
573 )
575 @overload
576 def cluster_addslotsrange(
577 self: SyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
578 ) -> bool: ...
580 @overload
581 def cluster_addslotsrange(
582 self: AsyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
583 ) -> Awaitable[bool]: ...
585 def cluster_addslotsrange(
586 self, target_node: "TargetNodesT", *slots: EncodableT
587 ) -> bool | Awaitable[bool]:
588 """
589 Similar to the CLUSTER ADDSLOTS command.
590 The difference between the two commands is that ADDSLOTS takes a list of slots
591 to assign to the node, while ADDSLOTSRANGE takes a list of slot ranges
592 (specified by start and end slots) to assign to the node.
594 :target_node: 'ClusterNode'
595 The node to execute the command on
597 For more information see https://redis.io/commands/cluster-addslotsrange
598 """
599 return self.execute_command(
600 "CLUSTER ADDSLOTSRANGE", *slots, target_nodes=target_node
601 )
603 @overload
604 def cluster_countkeysinslot(self: SyncClientProtocol, slot_id: int) -> int: ...
606 @overload
607 def cluster_countkeysinslot(
608 self: AsyncClientProtocol, slot_id: int
609 ) -> Awaitable[int]: ...
611 def cluster_countkeysinslot(self, slot_id: int) -> int | Awaitable[int]:
612 """
613 Return the number of local keys in the specified hash slot
614 Send to node based on specified slot_id
616 For more information see https://redis.io/commands/cluster-countkeysinslot
617 """
618 return self.execute_command("CLUSTER COUNTKEYSINSLOT", slot_id)
620 @overload
621 def cluster_count_failure_report(self: SyncClientProtocol, node_id: str) -> int: ...
623 @overload
624 def cluster_count_failure_report(
625 self: AsyncClientProtocol, node_id: str
626 ) -> Awaitable[int]: ...
628 def cluster_count_failure_report(self, node_id: str) -> int | Awaitable[int]:
629 """
630 Return the number of failure reports active for a given node
631 Sends to a random node
633 For more information see https://redis.io/commands/cluster-count-failure-reports
634 """
635 return self.execute_command("CLUSTER COUNT-FAILURE-REPORTS", node_id)
637 def cluster_delslots(self, *slots: EncodableT) -> List[bool]:
638 """
639 Set hash slots as unbound in the cluster.
640 It determines by it self what node the slot is in and sends it there
642 Returns a list of the results for each processed slot.
644 For more information see https://redis.io/commands/cluster-delslots
645 """
646 return [self.execute_command("CLUSTER DELSLOTS", slot) for slot in slots]
648 @overload
649 def cluster_delslotsrange(self: SyncClientProtocol, *slots: EncodableT) -> bool: ...
651 @overload
652 def cluster_delslotsrange(
653 self: AsyncClientProtocol, *slots: EncodableT
654 ) -> Awaitable[bool]: ...
656 def cluster_delslotsrange(self, *slots: EncodableT) -> bool | Awaitable[bool]:
657 """
658 Similar to the CLUSTER DELSLOTS command.
659 The difference is that CLUSTER DELSLOTS takes a list of hash slots to remove
660 from the node, while CLUSTER DELSLOTSRANGE takes a list of slot ranges to remove
661 from the node.
663 For more information see https://redis.io/commands/cluster-delslotsrange
664 """
665 return self.execute_command("CLUSTER DELSLOTSRANGE", *slots)
667 @overload
668 def cluster_failover(
669 self: SyncClientProtocol,
670 target_node: "TargetNodesT",
671 option: str | None = None,
672 ) -> bool: ...
674 @overload
675 def cluster_failover(
676 self: AsyncClientProtocol,
677 target_node: "TargetNodesT",
678 option: str | None = None,
679 ) -> Awaitable[bool]: ...
681 def cluster_failover(
682 self, target_node: "TargetNodesT", option: str | None = None
683 ) -> bool | Awaitable[bool]:
684 """
685 Forces a slave to perform a manual failover of its master
686 Sends to specified node
688 :target_node: 'ClusterNode'
689 The node to execute the command on
691 For more information see https://redis.io/commands/cluster-failover
692 """
693 if option:
694 if option.upper() not in ["FORCE", "TAKEOVER"]:
695 raise RedisError(
696 f"Invalid option for CLUSTER FAILOVER command: {option}"
697 )
698 else:
699 return self.execute_command(
700 "CLUSTER FAILOVER", option, target_nodes=target_node
701 )
702 else:
703 return self.execute_command("CLUSTER FAILOVER", target_nodes=target_node)
705 @overload
706 def cluster_info(
707 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
708 ) -> dict[str, str]: ...
710 @overload
711 def cluster_info(
712 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
713 ) -> Awaitable[dict[str, str]]: ...
715 def cluster_info(
716 self, target_nodes: "TargetNodesT" | None = None
717 ) -> dict[str, str] | Awaitable[dict[str, str]]:
718 """
719 Provides info about Redis Cluster node state.
720 The command will be sent to a random node in the cluster if no target
721 node is specified.
723 For more information see https://redis.io/commands/cluster-info
724 """
725 return self.execute_command("CLUSTER INFO", target_nodes=target_nodes)
727 @overload
728 def cluster_keyslot(self: SyncClientProtocol, key: str) -> int: ...
730 @overload
731 def cluster_keyslot(self: AsyncClientProtocol, key: str) -> Awaitable[int]: ...
733 def cluster_keyslot(self, key: str) -> int | Awaitable[int]:
734 """
735 Returns the hash slot of the specified key
736 Sends to random node in the cluster
738 For more information see https://redis.io/commands/cluster-keyslot
739 """
740 return self.execute_command("CLUSTER KEYSLOT", key)
742 @overload
743 def cluster_meet(
744 self: SyncClientProtocol,
745 host: str,
746 port: int,
747 target_nodes: "TargetNodesT" | None = None,
748 ) -> bool: ...
750 @overload
751 def cluster_meet(
752 self: AsyncClientProtocol,
753 host: str,
754 port: int,
755 target_nodes: "TargetNodesT" | None = None,
756 ) -> Awaitable[bool]: ...
758 def cluster_meet(
759 self, host: str, port: int, target_nodes: "TargetNodesT" | None = None
760 ) -> bool | Awaitable[bool]:
761 """
762 Force a node cluster to handshake with another node.
763 Sends to specified node.
765 For more information see https://redis.io/commands/cluster-meet
766 """
767 return self.execute_command(
768 "CLUSTER MEET", host, port, target_nodes=target_nodes
769 )
771 @overload
772 def cluster_nodes(self: SyncClientProtocol) -> dict[str, ClusterNodeDetail]: ...
774 @overload
775 def cluster_nodes(
776 self: AsyncClientProtocol,
777 ) -> Awaitable[dict[str, ClusterNodeDetail]]: ...
779 def cluster_nodes(
780 self,
781 ) -> dict[str, ClusterNodeDetail] | Awaitable[dict[str, ClusterNodeDetail]]:
782 """
783 Get Cluster config for the node.
784 Sends to random node in the cluster
786 For more information see https://redis.io/commands/cluster-nodes
787 """
788 return self.execute_command("CLUSTER NODES")
790 @overload
791 def cluster_replicate(
792 self: SyncClientProtocol, target_nodes: "TargetNodesT", node_id: str
793 ) -> bool: ...
795 @overload
796 def cluster_replicate(
797 self: AsyncClientProtocol, target_nodes: "TargetNodesT", node_id: str
798 ) -> Awaitable[bool]: ...
800 def cluster_replicate(
801 self, target_nodes: "TargetNodesT", node_id: str
802 ) -> bool | Awaitable[bool]:
803 """
804 Reconfigure a node as a slave of the specified master node
806 For more information see https://redis.io/commands/cluster-replicate
807 """
808 return self.execute_command(
809 "CLUSTER REPLICATE", node_id, target_nodes=target_nodes
810 )
812 @overload
813 def cluster_reset(
814 self: SyncClientProtocol,
815 soft: bool = True,
816 target_nodes: "TargetNodesT" | None = None,
817 ) -> bool: ...
819 @overload
820 def cluster_reset(
821 self: AsyncClientProtocol,
822 soft: bool = True,
823 target_nodes: "TargetNodesT" | None = None,
824 ) -> Awaitable[bool]: ...
826 def cluster_reset(
827 self, soft: bool = True, target_nodes: "TargetNodesT" | None = None
828 ) -> bool | Awaitable[bool]:
829 """
830 Reset a Redis Cluster node
832 If 'soft' is True then it will send 'SOFT' argument
833 If 'soft' is False then it will send 'HARD' argument
835 For more information see https://redis.io/commands/cluster-reset
836 """
837 return self.execute_command(
838 "CLUSTER RESET", b"SOFT" if soft else b"HARD", target_nodes=target_nodes
839 )
841 @overload
842 def cluster_save_config(
843 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
844 ) -> bool: ...
846 @overload
847 def cluster_save_config(
848 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
849 ) -> Awaitable[bool]: ...
851 def cluster_save_config(
852 self, target_nodes: "TargetNodesT" | None = None
853 ) -> bool | Awaitable[bool]:
854 """
855 Forces the node to save cluster state on disk
857 For more information see https://redis.io/commands/cluster-saveconfig
858 """
859 return self.execute_command("CLUSTER SAVECONFIG", target_nodes=target_nodes)
861 @overload
862 def cluster_get_keys_in_slot(
863 self: SyncClientProtocol, slot: int, num_keys: int
864 ) -> list[bytes | str]: ...
866 @overload
867 def cluster_get_keys_in_slot(
868 self: AsyncClientProtocol, slot: int, num_keys: int
869 ) -> Awaitable[list[bytes | str]]: ...
871 def cluster_get_keys_in_slot(
872 self, slot: int, num_keys: int
873 ) -> list[bytes | str] | Awaitable[list[bytes | str]]:
874 """
875 Returns the number of keys in the specified cluster slot
877 For more information see https://redis.io/commands/cluster-getkeysinslot
878 """
879 return self.execute_command("CLUSTER GETKEYSINSLOT", slot, num_keys)
881 @overload
882 def cluster_set_config_epoch(
883 self: SyncClientProtocol, epoch: int, target_nodes: "TargetNodesT" | None = None
884 ) -> bool: ...
886 @overload
887 def cluster_set_config_epoch(
888 self: AsyncClientProtocol,
889 epoch: int,
890 target_nodes: "TargetNodesT" | None = None,
891 ) -> Awaitable[bool]: ...
893 def cluster_set_config_epoch(
894 self, epoch: int, target_nodes: "TargetNodesT" | None = None
895 ) -> bool | Awaitable[bool]:
896 """
897 Set the configuration epoch in a new node
899 For more information see https://redis.io/commands/cluster-set-config-epoch
900 """
901 return self.execute_command(
902 "CLUSTER SET-CONFIG-EPOCH", epoch, target_nodes=target_nodes
903 )
905 @overload
906 def cluster_setslot(
907 self: SyncClientProtocol,
908 target_node: "TargetNodesT",
909 node_id: str,
910 slot_id: int,
911 state: str,
912 ) -> bool: ...
914 @overload
915 def cluster_setslot(
916 self: AsyncClientProtocol,
917 target_node: "TargetNodesT",
918 node_id: str,
919 slot_id: int,
920 state: str,
921 ) -> Awaitable[bool]: ...
923 def cluster_setslot(
924 self, target_node: "TargetNodesT", node_id: str, slot_id: int, state: str
925 ) -> bool | Awaitable[bool]:
926 """
927 Bind an hash slot to a specific node
929 :target_node: 'ClusterNode'
930 The node to execute the command on
932 For more information see https://redis.io/commands/cluster-setslot
933 """
934 if state.upper() in ("IMPORTING", "NODE", "MIGRATING"):
935 return self.execute_command(
936 "CLUSTER SETSLOT", slot_id, state, node_id, target_nodes=target_node
937 )
938 elif state.upper() == "STABLE":
939 raise RedisError('For "stable" state please use cluster_setslot_stable')
940 else:
941 raise RedisError(f"Invalid slot state: {state}")
943 @overload
944 def cluster_setslot_stable(self: SyncClientProtocol, slot_id: int) -> bool: ...
946 @overload
947 def cluster_setslot_stable(
948 self: AsyncClientProtocol, slot_id: int
949 ) -> Awaitable[bool]: ...
951 def cluster_setslot_stable(self, slot_id: int) -> bool | Awaitable[bool]:
952 """
953 Clears migrating / importing state from the slot.
954 It determines by it self what node the slot is in and sends it there.
956 For more information see https://redis.io/commands/cluster-setslot
957 """
958 return self.execute_command("CLUSTER SETSLOT", slot_id, "STABLE")
960 @overload
961 def cluster_replicas(
962 self: SyncClientProtocol,
963 node_id: str,
964 target_nodes: "TargetNodesT" | None = None,
965 ) -> dict[str, ClusterNodeDetail]: ...
967 @overload
968 def cluster_replicas(
969 self: AsyncClientProtocol,
970 node_id: str,
971 target_nodes: "TargetNodesT" | None = None,
972 ) -> Awaitable[dict[str, ClusterNodeDetail]]: ...
974 def cluster_replicas(
975 self, node_id: str, target_nodes: "TargetNodesT" | None = None
976 ) -> dict[str, ClusterNodeDetail] | Awaitable[dict[str, ClusterNodeDetail]]:
977 """
978 Provides a list of replica nodes replicating from the specified primary
979 target node.
981 For more information see https://redis.io/commands/cluster-replicas
982 """
983 return self.execute_command(
984 "CLUSTER REPLICAS", node_id, target_nodes=target_nodes
985 )
987 @overload
988 def cluster_slots(
989 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
990 ) -> list[Any]: ...
992 @overload
993 def cluster_slots(
994 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
995 ) -> Awaitable[list[Any]]: ...
997 def cluster_slots(
998 self, target_nodes: "TargetNodesT" | None = None
999 ) -> list[Any] | Awaitable[list[Any]]:
1000 """
1001 Get array of Cluster slot to node mappings
1003 For more information see https://redis.io/commands/cluster-slots
1004 """
1005 return self.execute_command("CLUSTER SLOTS", target_nodes=target_nodes)
1007 @overload
1008 def cluster_shards(
1009 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
1010 ) -> ClusterShardsResponse: ...
1012 @overload
1013 def cluster_shards(
1014 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
1015 ) -> Awaitable[ClusterShardsResponse]: ...
1017 def cluster_shards(
1018 self, target_nodes: "TargetNodesT" | None = None
1019 ) -> ClusterShardsResponse | Awaitable[ClusterShardsResponse]:
1020 """
1021 Returns details about the shards of the cluster.
1023 For more information see https://redis.io/commands/cluster-shards
1024 """
1025 return self.execute_command("CLUSTER SHARDS", target_nodes=target_nodes)
1027 @overload
1028 def cluster_myshardid(
1029 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
1030 ) -> bytes | str: ...
1032 @overload
1033 def cluster_myshardid(
1034 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
1035 ) -> Awaitable[bytes | str]: ...
1037 def cluster_myshardid(self, target_nodes: "TargetNodesT" | None = None) -> (
1038 bytes | str
1039 ) | Awaitable[bytes | str]:
1040 """
1041 Returns the shard ID of the node.
1043 For more information see https://redis.io/commands/cluster-myshardid/
1044 """
1045 return self.execute_command("CLUSTER MYSHARDID", target_nodes=target_nodes)
1047 @overload
1048 def cluster_links(
1049 self: SyncClientProtocol, target_node: "TargetNodesT"
1050 ) -> ClusterLinksResponse: ...
1052 @overload
1053 def cluster_links(
1054 self: AsyncClientProtocol, target_node: "TargetNodesT"
1055 ) -> Awaitable[ClusterLinksResponse]: ...
1057 def cluster_links(
1058 self, target_node: "TargetNodesT"
1059 ) -> ClusterLinksResponse | Awaitable[ClusterLinksResponse]:
1060 """
1061 Each node in a Redis Cluster maintains a pair of long-lived TCP link with each
1062 peer in the cluster: One for sending outbound messages towards the peer and one
1063 for receiving inbound messages from the peer.
1065 This command outputs information of all such peer links as an array.
1067 For more information see https://redis.io/commands/cluster-links
1068 """
1069 return self.execute_command("CLUSTER LINKS", target_nodes=target_node)
1071 def cluster_flushslots(self, target_nodes: "TargetNodesT" | None = None) -> None:
1072 raise NotImplementedError(
1073 "CLUSTER FLUSHSLOTS is intentionally not implemented in the client."
1074 )
1076 def cluster_bumpepoch(self, target_nodes: "TargetNodesT" | None = None) -> None:
1077 raise NotImplementedError(
1078 "CLUSTER BUMPEPOCH is intentionally not implemented in the client."
1079 )
1081 def readonly(self, target_nodes: "TargetNodesT" | None = None) -> ResponseT:
1082 """
1083 Enables read queries.
1084 The command will be sent to the default cluster node if target_nodes is
1085 not specified.
1087 For more information see https://redis.io/commands/readonly
1088 """
1089 if target_nodes == "replicas" or target_nodes == "all":
1090 # read_from_replicas will only be enabled if the READONLY command
1091 # is sent to all replicas
1092 self.read_from_replicas = True
1093 return self.execute_command("READONLY", target_nodes=target_nodes)
1095 def readwrite(self, target_nodes: "TargetNodesT" | None = None) -> ResponseT:
1096 """
1097 Disables read queries.
1098 The command will be sent to the default cluster node if target_nodes is
1099 not specified.
1101 For more information see https://redis.io/commands/readwrite
1102 """
1103 # Reset read from replicas flag
1104 self.read_from_replicas = False
1105 return self.execute_command("READWRITE", target_nodes=target_nodes)
1107 @deprecated_function(
1108 version="7.2.0",
1109 reason="Use client-side caching feature instead.",
1110 )
1111 def client_tracking_on(
1112 self,
1113 clientid: int | None = None,
1114 prefix: Sequence[KeyT] = [],
1115 bcast: bool = False,
1116 optin: bool = False,
1117 optout: bool = False,
1118 noloop: bool = False,
1119 target_nodes: "TargetNodesT" | None = "all",
1120 ) -> ResponseT:
1121 """
1122 Enables the tracking feature of the Redis server, that is used
1123 for server assisted client side caching.
1125 When clientid is provided - in target_nodes only the node that owns the
1126 connection with this id should be provided.
1127 When clientid is not provided - target_nodes can be any node.
1129 On a client with client-side caching enabled this raises ``RedisError``:
1130 the cache enables tracking itself.
1132 For more information see https://redis.io/commands/client-tracking
1133 """
1134 return self.client_tracking(
1135 True,
1136 clientid,
1137 prefix,
1138 bcast,
1139 optin,
1140 optout,
1141 noloop,
1142 target_nodes=target_nodes,
1143 )
1145 @deprecated_function(
1146 version="7.2.0",
1147 reason="Use client-side caching feature instead.",
1148 )
1149 def client_tracking_off(
1150 self,
1151 clientid: int | None = None,
1152 prefix: Sequence[KeyT] = [],
1153 bcast: bool = False,
1154 optin: bool = False,
1155 optout: bool = False,
1156 noloop: bool = False,
1157 target_nodes: "TargetNodesT" | None = "all",
1158 ) -> ResponseT:
1159 """
1160 Disables the tracking feature of the Redis server, that is used
1161 for server assisted client side caching.
1163 When clientid is provided - in target_nodes only the node that owns the
1164 connection with this id should be provided.
1165 When clientid is not provided - target_nodes can be any node.
1167 On a client with client-side caching enabled this raises ``RedisError``:
1168 turning tracking off would stop invalidations for cached replies.
1170 For more information see https://redis.io/commands/client-tracking
1171 """
1172 return self.client_tracking(
1173 False,
1174 clientid,
1175 prefix,
1176 bcast,
1177 optin,
1178 optout,
1179 noloop,
1180 target_nodes=target_nodes,
1181 )
1183 def hotkeys_start(
1184 self,
1185 metrics: List[HotkeysMetricsTypes],
1186 count: int | None = None,
1187 duration: int | None = None,
1188 sample_ratio: int | None = None,
1189 slots: List[int] | None = None,
1190 **kwargs,
1191 ) -> str | bytes:
1192 """
1193 Cluster client does not support hotkeys command. Please use the non-cluster client.
1195 For more information see https://redis.io/commands/hotkeys-start
1196 """
1197 raise NotImplementedError(
1198 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1199 )
1201 def hotkeys_stop(self, **kwargs) -> str | bytes:
1202 """
1203 Cluster client does not support hotkeys command. Please use the non-cluster client.
1205 For more information see https://redis.io/commands/hotkeys-stop
1206 """
1207 raise NotImplementedError(
1208 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1209 )
1211 def hotkeys_reset(self, **kwargs) -> str | bytes:
1212 """
1213 Cluster client does not support hotkeys command. Please use the non-cluster client.
1215 For more information see https://redis.io/commands/hotkeys-reset
1216 """
1217 raise NotImplementedError(
1218 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1219 )
1221 def hotkeys_get(self, **kwargs) -> list[dict[str | bytes, Any]]:
1222 """
1223 Cluster client does not support hotkeys command. Please use the non-cluster client.
1225 For more information see https://redis.io/commands/hotkeys-get
1226 """
1227 raise NotImplementedError(
1228 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1229 )
1232class AsyncClusterManagementCommands(
1233 ClusterManagementCommands, AsyncManagementCommands
1234):
1235 """
1236 A class for Redis Cluster management commands
1238 The class inherits from Redis's core ManagementCommands class and do the
1239 required adjustments to work with cluster mode
1240 """
1242 async def cluster_delslots(self, *slots: EncodableT) -> List[bool]:
1243 """
1244 Set hash slots as unbound in the cluster.
1245 It determines by it self what node the slot is in and sends it there
1247 Returns a list of the results for each processed slot.
1249 For more information see https://redis.io/commands/cluster-delslots
1250 """
1251 return await asyncio.gather(
1252 *(
1253 asyncio.create_task(self.execute_command("CLUSTER DELSLOTS", slot))
1254 for slot in slots
1255 )
1256 )
1258 @deprecated_function(
1259 version="7.2.0",
1260 reason="Use client-side caching feature instead.",
1261 )
1262 async def client_tracking_on(
1263 self,
1264 clientid: int | None = None,
1265 prefix: Sequence[KeyT] = [],
1266 bcast: bool = False,
1267 optin: bool = False,
1268 optout: bool = False,
1269 noloop: bool = False,
1270 target_nodes: "TargetNodesT" | None = "all",
1271 ) -> ResponseT:
1272 """
1273 Enables the tracking feature of the Redis server, that is used
1274 for server assisted client side caching.
1276 When clientid is provided - in target_nodes only the node that owns the
1277 connection with this id should be provided.
1278 When clientid is not provided - target_nodes can be any node.
1280 For more information see https://redis.io/commands/client-tracking
1281 """
1282 return await self.client_tracking(
1283 True,
1284 clientid,
1285 prefix,
1286 bcast,
1287 optin,
1288 optout,
1289 noloop,
1290 target_nodes=target_nodes,
1291 )
1293 @deprecated_function(
1294 version="7.2.0",
1295 reason="Use client-side caching feature instead.",
1296 )
1297 async def client_tracking_off(
1298 self,
1299 clientid: int | None = None,
1300 prefix: Sequence[KeyT] = [],
1301 bcast: bool = False,
1302 optin: bool = False,
1303 optout: bool = False,
1304 noloop: bool = False,
1305 target_nodes: "TargetNodesT" | None = "all",
1306 ) -> ResponseT:
1307 """
1308 Disables the tracking feature of the Redis server, that is used
1309 for server assisted client side caching.
1311 When clientid is provided - in target_nodes only the node that owns the
1312 connection with this id should be provided.
1313 When clientid is not provided - target_nodes can be any node.
1315 For more information see https://redis.io/commands/client-tracking
1316 """
1317 return await self.client_tracking(
1318 False,
1319 clientid,
1320 prefix,
1321 bcast,
1322 optin,
1323 optout,
1324 noloop,
1325 target_nodes=target_nodes,
1326 )
1328 async def hotkeys_start(
1329 self,
1330 metrics: List[HotkeysMetricsTypes],
1331 count: int | None = None,
1332 duration: int | None = None,
1333 sample_ratio: int | None = None,
1334 slots: List[int] | None = None,
1335 **kwargs,
1336 ) -> Awaitable[str | bytes]:
1337 """
1338 Cluster client does not support hotkeys command. Please use the non-cluster client.
1340 For more information see https://redis.io/commands/hotkeys-start
1341 """
1342 raise NotImplementedError(
1343 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1344 )
1346 async def hotkeys_stop(self, **kwargs) -> Awaitable[str | bytes]:
1347 """
1348 Cluster client does not support hotkeys command. Please use the non-cluster client.
1350 For more information see https://redis.io/commands/hotkeys-stop
1351 """
1352 raise NotImplementedError(
1353 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1354 )
1356 async def hotkeys_reset(self, **kwargs) -> Awaitable[str | bytes]:
1357 """
1358 Cluster client does not support hotkeys command. Please use the non-cluster client.
1360 For more information see https://redis.io/commands/hotkeys-reset
1361 """
1362 raise NotImplementedError(
1363 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1364 )
1366 async def hotkeys_get(self, **kwargs) -> Awaitable[list[dict[str | bytes, Any]]]:
1367 """
1368 Cluster client does not support hotkeys command. Please use the non-cluster client.
1370 For more information see https://redis.io/commands/hotkeys-get
1371 """
1372 raise NotImplementedError(
1373 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1374 )
1377class ClusterDataAccessCommands(DataAccessCommands):
1378 """
1379 A class for Redis Cluster Data Access Commands
1381 The class inherits from Redis's core DataAccessCommand class and do the
1382 required adjustments to work with cluster mode
1383 """
1385 @overload
1386 def stralgo(
1387 self: SyncClientProtocol,
1388 algo: Literal["LCS"],
1389 value1: KeyT,
1390 value2: KeyT,
1391 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1392 len: bool = False,
1393 idx: bool = False,
1394 minmatchlen: int | None = None,
1395 withmatchlen: bool = False,
1396 **kwargs,
1397 ) -> StralgoResponse: ...
1399 @overload
1400 def stralgo(
1401 self: AsyncClientProtocol,
1402 algo: Literal["LCS"],
1403 value1: KeyT,
1404 value2: KeyT,
1405 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1406 len: bool = False,
1407 idx: bool = False,
1408 minmatchlen: int | None = None,
1409 withmatchlen: bool = False,
1410 **kwargs,
1411 ) -> Awaitable[StralgoResponse]: ...
1413 def stralgo(
1414 self,
1415 algo: Literal["LCS"],
1416 value1: KeyT,
1417 value2: KeyT,
1418 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1419 len: bool = False,
1420 idx: bool = False,
1421 minmatchlen: int | None = None,
1422 withmatchlen: bool = False,
1423 **kwargs,
1424 ) -> StralgoResponse | Awaitable[StralgoResponse]:
1425 """
1426 Implements complex algorithms that operate on strings.
1427 Right now the only algorithm implemented is the LCS algorithm
1428 (longest common substring). However new algorithms could be
1429 implemented in the future.
1431 ``algo`` Right now must be LCS
1432 ``value1`` and ``value2`` Can be two strings or two keys
1433 ``specific_argument`` Specifying if the arguments to the algorithm
1434 will be keys or strings. strings is the default.
1435 ``len`` Returns just the len of the match.
1436 ``idx`` Returns the match positions in each string.
1437 ``minmatchlen`` Restrict the list of matches to the ones of a given
1438 minimal length. Can be provided only when ``idx`` set to True.
1439 ``withmatchlen`` Returns the matches with the len of the match.
1440 Can be provided only when ``idx`` set to True.
1442 For more information see https://redis.io/commands/stralgo
1443 """
1444 target_nodes = kwargs.pop("target_nodes", None)
1445 if specific_argument == "strings" and target_nodes is None:
1446 target_nodes = "default-node"
1447 kwargs.update({"target_nodes": target_nodes})
1448 return super().stralgo(
1449 algo,
1450 value1,
1451 value2,
1452 specific_argument,
1453 len,
1454 idx,
1455 minmatchlen,
1456 withmatchlen,
1457 **kwargs,
1458 )
1460 def scan_iter(
1461 self,
1462 match: PatternT | None = None,
1463 count: int | None = None,
1464 _type: str | None = None,
1465 **kwargs,
1466 ) -> Iterator:
1467 # Do the first query with cursor=0 for all nodes
1468 cursors, data = self.scan(match=match, count=count, _type=_type, **kwargs)
1469 yield from data
1471 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1472 if cursors:
1473 # Get nodes by name
1474 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1476 # Iterate over each node till its cursor is 0
1477 kwargs.pop("target_nodes", None)
1478 while cursors:
1479 for name, cursor in cursors.items():
1480 cur, data = self.scan(
1481 cursor=cursor,
1482 match=match,
1483 count=count,
1484 _type=_type,
1485 target_nodes=nodes[name],
1486 **kwargs,
1487 )
1488 yield from data
1489 cursors[name] = cur[name]
1491 cursors = {
1492 name: cursor for name, cursor in cursors.items() if cursor != 0
1493 }
1496class AsyncClusterDataAccessCommands(
1497 ClusterDataAccessCommands, AsyncDataAccessCommands
1498):
1499 """
1500 A class for Redis Cluster Data Access Commands
1502 The class inherits from Redis's core DataAccessCommand class and do the
1503 required adjustments to work with cluster mode
1504 """
1506 async def scan_iter(
1507 self,
1508 match: PatternT | None = None,
1509 count: int | None = None,
1510 _type: str | None = None,
1511 **kwargs,
1512 ) -> AsyncIterator:
1513 # Do the first query with cursor=0 for all nodes
1514 cursors, data = await self.scan(match=match, count=count, _type=_type, **kwargs)
1515 for value in data:
1516 yield value
1518 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1519 if cursors:
1520 # Get nodes by name
1521 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1523 # Iterate over each node till its cursor is 0
1524 kwargs.pop("target_nodes", None)
1525 while cursors:
1526 for name, cursor in cursors.items():
1527 cur, data = await self.scan(
1528 cursor=cursor,
1529 match=match,
1530 count=count,
1531 _type=_type,
1532 target_nodes=nodes[name],
1533 **kwargs,
1534 )
1535 for value in data:
1536 yield value
1537 cursors[name] = cur[name]
1539 cursors = {
1540 name: cursor for name, cursor in cursors.items() if cursor != 0
1541 }
1544class ClusterBlessCommands(BlessCommands):
1545 """
1546 A class for Redis Cluster BLESS commands
1548 The class inherits from Redis's core BlessCommands class and does the
1549 required adjustments to work with cluster mode
1550 """
1552 def bless_scan_iter(
1553 self,
1554 flag: BlessFlag,
1555 count: int | None = None,
1556 **kwargs,
1557 ) -> Iterator[bytes | str]:
1558 # Do the first query with cursor=0 for all nodes
1559 cursors, data = self.bless_scan(cursor=0, flag=flag, count=count, **kwargs)
1560 yield from data
1562 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1563 if cursors:
1564 # Get nodes by name
1565 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1567 # Iterate over each node till its cursor is 0
1568 kwargs.pop("target_nodes", None)
1569 while cursors:
1570 for name, cursor in cursors.items():
1571 cur, data = self.bless_scan(
1572 cursor=cursor,
1573 flag=flag,
1574 count=count,
1575 target_nodes=nodes[name],
1576 **kwargs,
1577 )
1578 yield from data
1579 cursors[name] = cur[name]
1581 cursors = {
1582 name: cursor for name, cursor in cursors.items() if cursor != 0
1583 }
1586class AsyncClusterBlessCommands(ClusterBlessCommands, AsyncBlessCommands):
1587 """
1588 A class for Redis Cluster BLESS commands
1590 The class inherits from Redis's core BlessCommands class and does the
1591 required adjustments to work with cluster mode
1592 """
1594 async def bless_scan_iter(
1595 self,
1596 flag: BlessFlag,
1597 count: int | None = None,
1598 **kwargs,
1599 ) -> AsyncIterator[bytes | str]:
1600 # Do the first query with cursor=0 for all nodes
1601 cursors, data = await self.bless_scan(
1602 cursor=0, flag=flag, count=count, **kwargs
1603 )
1604 for value in data:
1605 yield value
1607 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1608 if cursors:
1609 # Get nodes by name
1610 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1612 # Iterate over each node till its cursor is 0
1613 kwargs.pop("target_nodes", None)
1614 while cursors:
1615 for name, cursor in cursors.items():
1616 cur, data = await self.bless_scan(
1617 cursor=cursor,
1618 flag=flag,
1619 count=count,
1620 target_nodes=nodes[name],
1621 **kwargs,
1622 )
1623 for value in data:
1624 yield value
1625 cursors[name] = cur[name]
1627 cursors = {
1628 name: cursor for name, cursor in cursors.items() if cursor != 0
1629 }
1632class RedisClusterCommands(
1633 ClusterMultiKeyCommands,
1634 ClusterManagementCommands,
1635 ACLCommands,
1636 PubSubCommands,
1637 ClusterDataAccessCommands,
1638 ScriptCommands,
1639 FunctionCommands,
1640 ClusterBlessCommands,
1641 ModuleCommands,
1642 RedisModuleCommands,
1643):
1644 """
1645 A class for all Redis Cluster commands
1647 For key-based commands, the target node(s) will be internally determined
1648 by the keys' hash slot.
1649 Non-key-based commands can be executed with the 'target_nodes' argument to
1650 target specific nodes. By default, if target_nodes is not specified, the
1651 command will be executed on the default cluster node.
1653 :param target_nodes: type can be one of the following:
1654 - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
1655 - 'ClusterNode'
1656 - 'list(ClusterNodes)'
1657 - 'dict(any:clusterNodes)'
1659 for example:
1660 r.cluster_info(target_nodes=RedisCluster.ALL_NODES)
1661 """
1664class AsyncRedisClusterCommands(
1665 AsyncClusterMultiKeyCommands,
1666 AsyncClusterManagementCommands,
1667 AsyncACLCommands,
1668 PubSubCommands,
1669 AsyncClusterDataAccessCommands,
1670 AsyncScriptCommands,
1671 AsyncFunctionCommands,
1672 AsyncClusterBlessCommands,
1673 AsyncModuleCommands,
1674 AsyncRedisModuleCommands,
1675):
1676 """
1677 A class for all Redis Cluster commands
1679 For key-based commands, the target node(s) will be internally determined
1680 by the keys' hash slot.
1681 Non-key-based commands can be executed with the 'target_nodes' argument to
1682 target specific nodes. By default, if target_nodes is not specified, the
1683 command will be executed on the default cluster node.
1685 :param target_nodes: type can be one of the following:
1686 - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
1687 - 'ClusterNode'
1688 - 'list(ClusterNodes)'
1689 - 'dict(any:clusterNodes)'
1691 for example:
1692 r.cluster_info(target_nodes=RedisCluster.ALL_NODES)
1693 """