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
4from typing import (
5 TYPE_CHECKING,
6 Any,
7 AsyncIterator,
8 Awaitable,
9 Dict,
10 Iterable,
11 Iterator,
12 List,
13 Literal,
14 Mapping,
15 NoReturn,
16 Sequence,
17 overload,
18)
20from redis.crc import key_slot
21from redis.exceptions import RedisClusterException, RedisError
22from redis.typing import (
23 AnyKeyT,
24 AsyncClientProtocol,
25 ClusterCommandsProtocol,
26 ClusterLinksResponse,
27 ClusterNodeDetail,
28 ClusterShardsResponse,
29 EncodableT,
30 KeysT,
31 KeyT,
32 PatternT,
33 ResponseT,
34 StralgoResponse,
35 SyncClientProtocol,
36)
37from redis.utils import deprecated_function
39from .core import (
40 ACLCommands,
41 AsyncACLCommands,
42 AsyncDataAccessCommands,
43 AsyncFunctionCommands,
44 AsyncManagementCommands,
45 AsyncModuleCommands,
46 AsyncScriptCommands,
47 DataAccessCommands,
48 FunctionCommands,
49 HotkeysMetricsTypes,
50 ManagementCommands,
51 ModuleCommands,
52 PubSubCommands,
53 ScriptCommands,
54)
55from .helpers import list_or_args
56from .redismodules import AsyncRedisModuleCommands, RedisModuleCommands
58if TYPE_CHECKING:
59 from redis.asyncio.cluster import TargetNodesT
61# Not complete, but covers the major ones
62# https://redis.io/commands
63READ_COMMANDS = frozenset(
64 [
65 # Bit Operations
66 "BITCOUNT",
67 "BITFIELD_RO",
68 "BITPOS",
69 # Scripting
70 "EVAL_RO",
71 "EVALSHA_RO",
72 "FCALL_RO",
73 # Key Operations
74 "DBSIZE",
75 "DIGEST",
76 "DUMP",
77 "EXISTS",
78 "EXPIRETIME",
79 "PEXPIRETIME",
80 "KEYS",
81 "SCAN",
82 "PTTL",
83 "RANDOMKEY",
84 "TTL",
85 "TYPE",
86 # String Operations
87 "GET",
88 "GETBIT",
89 "GETRANGE",
90 "MGET",
91 "STRLEN",
92 "LCS",
93 # Geo Operations
94 "GEODIST",
95 "GEOHASH",
96 "GEOPOS",
97 "GEOSEARCH",
98 # Hash Operations
99 "HEXISTS",
100 "HGET",
101 "HGETALL",
102 "HKEYS",
103 "HLEN",
104 "HMGET",
105 "HSTRLEN",
106 "HVALS",
107 "HRANDFIELD",
108 "HEXPIRETIME",
109 "HPEXPIRETIME",
110 "HTTL",
111 "HPTTL",
112 "HSCAN",
113 # List Operations
114 "LINDEX",
115 "LPOS",
116 "LLEN",
117 "LRANGE",
118 # Set Operations
119 "SCARD",
120 "SDIFF",
121 "SDIFFCARD",
122 "SINTER",
123 "SINTERCARD",
124 "SISMEMBER",
125 "SMISMEMBER",
126 "SMEMBERS",
127 "SRANDMEMBER",
128 "SUNION",
129 "SUNIONCARD",
130 "SSCAN",
131 # Sorted Set Operations
132 "ZCARD",
133 "ZCOUNT",
134 "ZDIFF",
135 "ZINTER",
136 "ZINTERCARD",
137 "ZLEXCOUNT",
138 "ZMSCORE",
139 "ZRANDMEMBER",
140 "ZRANGE",
141 "ZRANGEBYLEX",
142 "ZRANGEBYSCORE",
143 "ZRANK",
144 "ZREVRANGE",
145 "ZREVRANGEBYLEX",
146 "ZREVRANGEBYSCORE",
147 "ZREVRANK",
148 "ZSCAN",
149 "ZSCORE",
150 "ZUNION",
151 # Stream Operations
152 "XLEN",
153 "XPENDING",
154 "XRANGE",
155 "XREAD",
156 "XREVRANGE",
157 # JSON Module
158 "JSON.ARRINDEX",
159 "JSON.ARRLEN",
160 "JSON.GET",
161 "JSON.MGET",
162 "JSON.OBJKEYS",
163 "JSON.OBJLEN",
164 "JSON.RESP",
165 "JSON.STRLEN",
166 "JSON.TYPE",
167 # RediSearch Module
168 "FT.ALIASLIST",
169 "FT.EXPLAIN",
170 "FT.INFO",
171 "FT.PROFILE",
172 "FT.SEARCH",
173 ]
174)
177class ClusterMultiKeyCommands(ClusterCommandsProtocol):
178 """
179 A class containing commands that handle more than one key
180 """
182 def _partition_keys_by_slot(self, keys: Iterable[KeyT]) -> Dict[int, List[KeyT]]:
183 """Split keys into a dictionary that maps a slot to a list of keys."""
185 slots_to_keys = {}
186 for key in keys:
187 slot = key_slot(self.encoder.encode(key))
188 slots_to_keys.setdefault(slot, []).append(key)
190 return slots_to_keys
192 def _partition_pairs_by_slot(
193 self, mapping: Mapping[AnyKeyT, EncodableT]
194 ) -> Dict[int, List[EncodableT]]:
195 """Split pairs into a dictionary that maps a slot to a list of pairs."""
197 slots_to_pairs = {}
198 for pair in mapping.items():
199 slot = key_slot(self.encoder.encode(pair[0]))
200 slots_to_pairs.setdefault(slot, []).extend(pair)
202 return slots_to_pairs
204 def _execute_pipeline_by_slot(
205 self, command: str, slots_to_args: Mapping[int, Iterable[EncodableT]]
206 ) -> List[Any]:
207 read_from_replicas = self.read_from_replicas and command in READ_COMMANDS
208 pipe = self.pipeline()
209 [
210 pipe.execute_command(
211 command,
212 *slot_args,
213 target_nodes=[
214 self.nodes_manager.get_node_from_slot(slot, read_from_replicas)
215 ],
216 )
217 for slot, slot_args in slots_to_args.items()
218 ]
219 return pipe.execute()
221 def _reorder_keys_by_command(
222 self,
223 keys: Iterable[KeyT],
224 slots_to_args: Mapping[int, Iterable[EncodableT]],
225 responses: Iterable[Any],
226 ) -> List[Any]:
227 results = {
228 k: v
229 for slot_values, response in zip(slots_to_args.values(), responses)
230 for k, v in zip(slot_values, response)
231 }
232 return [results[key] for key in keys]
234 def mget_nonatomic(self, keys: KeysT, *args: KeyT) -> List[Any | None]:
235 """
236 Splits the keys into different slots and then calls MGET
237 for the keys of every slot. This operation will not be atomic
238 if keys belong to more than one slot.
240 Returns a list of values ordered identically to ``keys``
242 For more information see https://redis.io/commands/mget
243 """
245 # Concatenate all keys into a list
246 keys = list_or_args(keys, args)
248 # Split keys into slots
249 slots_to_keys = self._partition_keys_by_slot(keys)
251 # Execute commands using a pipeline
252 res = self._execute_pipeline_by_slot("MGET", slots_to_keys)
254 # Reorder keys in the order the user provided & return
255 return self._reorder_keys_by_command(keys, slots_to_keys, res)
257 def mset_nonatomic(self, mapping: Mapping[AnyKeyT, EncodableT]) -> List[bool]:
258 """
259 Sets key/values based on a mapping. Mapping is a dictionary of
260 key/value pairs. Both keys and values should be strings or types that
261 can be cast to a string via str().
263 Splits the keys into different slots and then calls MSET
264 for the keys of every slot. This operation will not be atomic
265 if keys belong to more than one slot.
267 For more information see https://redis.io/commands/mset
268 """
270 # Partition the keys by slot
271 slots_to_pairs = self._partition_pairs_by_slot(mapping)
273 # Execute commands using a pipeline & return list of replies
274 return self._execute_pipeline_by_slot("MSET", slots_to_pairs)
276 def _split_command_across_slots(self, command: str, *keys: KeyT) -> int:
277 """
278 Runs the given command once for the keys
279 of each slot. Returns the sum of the return values.
280 """
282 # Partition the keys by slot
283 slots_to_keys = self._partition_keys_by_slot(keys)
285 # Sum up the reply from each command
286 return sum(self._execute_pipeline_by_slot(command, slots_to_keys))
288 @overload
289 def exists(self: SyncClientProtocol, *keys: KeyT) -> int: ...
291 @overload
292 def exists(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
294 def exists(self, *keys: KeyT) -> int | Awaitable[int]:
295 """
296 Returns the number of ``names`` that exist in the
297 whole cluster. The keys are first split up into slots
298 and then an EXISTS command is sent for every slot
300 For more information see https://redis.io/commands/exists
301 """
302 return self._split_command_across_slots("EXISTS", *keys)
304 @overload
305 def delete(self: SyncClientProtocol, *keys: KeyT) -> int: ...
307 @overload
308 def delete(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
310 def delete(self, *keys: KeyT) -> int | Awaitable[int]:
311 """
312 Deletes the given keys in the cluster.
313 The keys are first split up into slots
314 and then an DEL command is sent for every slot
316 Non-existent keys are ignored.
317 Returns the number of keys that were deleted.
319 For more information see https://redis.io/commands/del
320 """
321 return self._split_command_across_slots("DEL", *keys)
323 @overload
324 def touch(self: SyncClientProtocol, *keys: KeyT) -> int: ...
326 @overload
327 def touch(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
329 def touch(self, *keys: KeyT) -> int | Awaitable[int]:
330 """
331 Updates the last access time of given keys across the
332 cluster.
334 The keys are first split up into slots
335 and then an TOUCH command is sent for every slot
337 Non-existent keys are ignored.
338 Returns the number of keys that were touched.
340 For more information see https://redis.io/commands/touch
341 """
342 return self._split_command_across_slots("TOUCH", *keys)
344 @overload
345 def unlink(self: SyncClientProtocol, *keys: KeyT) -> int: ...
347 @overload
348 def unlink(self: AsyncClientProtocol, *keys: KeyT) -> Awaitable[int]: ...
350 def unlink(self, *keys: KeyT) -> int | Awaitable[int]:
351 """
352 Remove the specified keys in a different thread.
354 The keys are first split up into slots
355 and then an TOUCH command is sent for every slot
357 Non-existent keys are ignored.
358 Returns the number of keys that were unlinked.
360 For more information see https://redis.io/commands/unlink
361 """
362 return self._split_command_across_slots("UNLINK", *keys)
365class AsyncClusterMultiKeyCommands(ClusterMultiKeyCommands):
366 """
367 A class containing commands that handle more than one key
368 """
370 async def mget_nonatomic(self, keys: KeysT, *args: KeyT) -> List[Any | None]:
371 """
372 Splits the keys into different slots and then calls MGET
373 for the keys of every slot. This operation will not be atomic
374 if keys belong to more than one slot.
376 Returns a list of values ordered identically to ``keys``
378 For more information see https://redis.io/commands/mget
379 """
381 # Concatenate all keys into a list
382 keys = list_or_args(keys, args)
384 # Split keys into slots
385 slots_to_keys = self._partition_keys_by_slot(keys)
387 # Execute commands using a pipeline
388 res = await self._execute_pipeline_by_slot("MGET", slots_to_keys)
390 # Reorder keys in the order the user provided & return
391 return self._reorder_keys_by_command(keys, slots_to_keys, res)
393 async def mset_nonatomic(self, mapping: Mapping[AnyKeyT, EncodableT]) -> List[bool]:
394 """
395 Sets key/values based on a mapping. Mapping is a dictionary of
396 key/value pairs. Both keys and values should be strings or types that
397 can be cast to a string via str().
399 Splits the keys into different slots and then calls MSET
400 for the keys of every slot. This operation will not be atomic
401 if keys belong to more than one slot.
403 For more information see https://redis.io/commands/mset
404 """
406 # Partition the keys by slot
407 slots_to_pairs = self._partition_pairs_by_slot(mapping)
409 # Execute commands using a pipeline & return list of replies
410 return await self._execute_pipeline_by_slot("MSET", slots_to_pairs)
412 async def _split_command_across_slots(self, command: str, *keys: KeyT) -> int:
413 """
414 Runs the given command once for the keys
415 of each slot. Returns the sum of the return values.
416 """
418 # Partition the keys by slot
419 slots_to_keys = self._partition_keys_by_slot(keys)
421 # Sum up the reply from each command
422 return sum(await self._execute_pipeline_by_slot(command, slots_to_keys))
424 async def _execute_pipeline_by_slot(
425 self, command: str, slots_to_args: Mapping[int, Iterable[EncodableT]]
426 ) -> List[Any]:
427 if self._initialize:
428 await self.initialize()
429 read_from_replicas = self.read_from_replicas and command in READ_COMMANDS
430 pipe = self.pipeline()
431 [
432 pipe.execute_command(
433 command,
434 *slot_args,
435 target_nodes=[
436 self.nodes_manager.get_node_from_slot(slot, read_from_replicas)
437 ],
438 )
439 for slot, slot_args in slots_to_args.items()
440 ]
441 return await pipe.execute()
444class ClusterManagementCommands(ManagementCommands):
445 """
446 A class for Redis Cluster management commands
448 The class inherits from Redis's core ManagementCommands class and do the
449 required adjustments to work with cluster mode
450 """
452 def slaveof(self, *args, **kwargs) -> NoReturn:
453 """
454 Make the server a replica of another instance, or promote it as master.
456 For more information see https://redis.io/commands/slaveof
457 """
458 raise RedisClusterException("SLAVEOF is not supported in cluster mode")
460 def replicaof(self, *args, **kwargs) -> NoReturn:
461 """
462 Make the server a replica of another instance, or promote it as master.
464 For more information see https://redis.io/commands/replicaof
465 """
466 raise RedisClusterException("REPLICAOF is not supported in cluster mode")
468 def swapdb(self, *args, **kwargs) -> NoReturn:
469 """
470 Swaps two Redis databases.
472 For more information see https://redis.io/commands/swapdb
473 """
474 raise RedisClusterException("SWAPDB is not supported in cluster mode")
476 @overload
477 def cluster_myid(
478 self: SyncClientProtocol, target_node: "TargetNodesT"
479 ) -> bytes | str: ...
481 @overload
482 def cluster_myid(
483 self: AsyncClientProtocol, target_node: "TargetNodesT"
484 ) -> Awaitable[bytes | str]: ...
486 def cluster_myid(self, target_node: "TargetNodesT") -> (bytes | str) | Awaitable[
487 bytes | str
488 ]:
489 """
490 Returns the node's id.
492 :target_node: 'ClusterNode'
493 The node to execute the command on
495 For more information check https://redis.io/commands/cluster-myid/
496 """
497 return self.execute_command("CLUSTER MYID", target_nodes=target_node)
499 @overload
500 def cluster_addslots(
501 self: SyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
502 ) -> bool: ...
504 @overload
505 def cluster_addslots(
506 self: AsyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
507 ) -> Awaitable[bool]: ...
509 def cluster_addslots(
510 self, target_node: "TargetNodesT", *slots: EncodableT
511 ) -> bool | Awaitable[bool]:
512 """
513 Assign new hash slots to receiving node. Sends to specified node.
515 :target_node: 'ClusterNode'
516 The node to execute the command on
518 For more information see https://redis.io/commands/cluster-addslots
519 """
520 return self.execute_command(
521 "CLUSTER ADDSLOTS", *slots, target_nodes=target_node
522 )
524 @overload
525 def cluster_addslotsrange(
526 self: SyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
527 ) -> bool: ...
529 @overload
530 def cluster_addslotsrange(
531 self: AsyncClientProtocol, target_node: "TargetNodesT", *slots: EncodableT
532 ) -> Awaitable[bool]: ...
534 def cluster_addslotsrange(
535 self, target_node: "TargetNodesT", *slots: EncodableT
536 ) -> bool | Awaitable[bool]:
537 """
538 Similar to the CLUSTER ADDSLOTS command.
539 The difference between the two commands is that ADDSLOTS takes a list of slots
540 to assign to the node, while ADDSLOTSRANGE takes a list of slot ranges
541 (specified by start and end slots) to assign to the node.
543 :target_node: 'ClusterNode'
544 The node to execute the command on
546 For more information see https://redis.io/commands/cluster-addslotsrange
547 """
548 return self.execute_command(
549 "CLUSTER ADDSLOTSRANGE", *slots, target_nodes=target_node
550 )
552 @overload
553 def cluster_countkeysinslot(self: SyncClientProtocol, slot_id: int) -> int: ...
555 @overload
556 def cluster_countkeysinslot(
557 self: AsyncClientProtocol, slot_id: int
558 ) -> Awaitable[int]: ...
560 def cluster_countkeysinslot(self, slot_id: int) -> int | Awaitable[int]:
561 """
562 Return the number of local keys in the specified hash slot
563 Send to node based on specified slot_id
565 For more information see https://redis.io/commands/cluster-countkeysinslot
566 """
567 return self.execute_command("CLUSTER COUNTKEYSINSLOT", slot_id)
569 @overload
570 def cluster_count_failure_report(self: SyncClientProtocol, node_id: str) -> int: ...
572 @overload
573 def cluster_count_failure_report(
574 self: AsyncClientProtocol, node_id: str
575 ) -> Awaitable[int]: ...
577 def cluster_count_failure_report(self, node_id: str) -> int | Awaitable[int]:
578 """
579 Return the number of failure reports active for a given node
580 Sends to a random node
582 For more information see https://redis.io/commands/cluster-count-failure-reports
583 """
584 return self.execute_command("CLUSTER COUNT-FAILURE-REPORTS", node_id)
586 def cluster_delslots(self, *slots: EncodableT) -> List[bool]:
587 """
588 Set hash slots as unbound in the cluster.
589 It determines by it self what node the slot is in and sends it there
591 Returns a list of the results for each processed slot.
593 For more information see https://redis.io/commands/cluster-delslots
594 """
595 return [self.execute_command("CLUSTER DELSLOTS", slot) for slot in slots]
597 @overload
598 def cluster_delslotsrange(self: SyncClientProtocol, *slots: EncodableT) -> bool: ...
600 @overload
601 def cluster_delslotsrange(
602 self: AsyncClientProtocol, *slots: EncodableT
603 ) -> Awaitable[bool]: ...
605 def cluster_delslotsrange(self, *slots: EncodableT) -> bool | Awaitable[bool]:
606 """
607 Similar to the CLUSTER DELSLOTS command.
608 The difference is that CLUSTER DELSLOTS takes a list of hash slots to remove
609 from the node, while CLUSTER DELSLOTSRANGE takes a list of slot ranges to remove
610 from the node.
612 For more information see https://redis.io/commands/cluster-delslotsrange
613 """
614 return self.execute_command("CLUSTER DELSLOTSRANGE", *slots)
616 @overload
617 def cluster_failover(
618 self: SyncClientProtocol,
619 target_node: "TargetNodesT",
620 option: str | None = None,
621 ) -> bool: ...
623 @overload
624 def cluster_failover(
625 self: AsyncClientProtocol,
626 target_node: "TargetNodesT",
627 option: str | None = None,
628 ) -> Awaitable[bool]: ...
630 def cluster_failover(
631 self, target_node: "TargetNodesT", option: str | None = None
632 ) -> bool | Awaitable[bool]:
633 """
634 Forces a slave to perform a manual failover of its master
635 Sends to specified node
637 :target_node: 'ClusterNode'
638 The node to execute the command on
640 For more information see https://redis.io/commands/cluster-failover
641 """
642 if option:
643 if option.upper() not in ["FORCE", "TAKEOVER"]:
644 raise RedisError(
645 f"Invalid option for CLUSTER FAILOVER command: {option}"
646 )
647 else:
648 return self.execute_command(
649 "CLUSTER FAILOVER", option, target_nodes=target_node
650 )
651 else:
652 return self.execute_command("CLUSTER FAILOVER", target_nodes=target_node)
654 @overload
655 def cluster_info(
656 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
657 ) -> dict[str, str]: ...
659 @overload
660 def cluster_info(
661 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
662 ) -> Awaitable[dict[str, str]]: ...
664 def cluster_info(
665 self, target_nodes: "TargetNodesT" | None = None
666 ) -> dict[str, str] | Awaitable[dict[str, str]]:
667 """
668 Provides info about Redis Cluster node state.
669 The command will be sent to a random node in the cluster if no target
670 node is specified.
672 For more information see https://redis.io/commands/cluster-info
673 """
674 return self.execute_command("CLUSTER INFO", target_nodes=target_nodes)
676 @overload
677 def cluster_keyslot(self: SyncClientProtocol, key: str) -> int: ...
679 @overload
680 def cluster_keyslot(self: AsyncClientProtocol, key: str) -> Awaitable[int]: ...
682 def cluster_keyslot(self, key: str) -> int | Awaitable[int]:
683 """
684 Returns the hash slot of the specified key
685 Sends to random node in the cluster
687 For more information see https://redis.io/commands/cluster-keyslot
688 """
689 return self.execute_command("CLUSTER KEYSLOT", key)
691 @overload
692 def cluster_meet(
693 self: SyncClientProtocol,
694 host: str,
695 port: int,
696 target_nodes: "TargetNodesT" | None = None,
697 ) -> bool: ...
699 @overload
700 def cluster_meet(
701 self: AsyncClientProtocol,
702 host: str,
703 port: int,
704 target_nodes: "TargetNodesT" | None = None,
705 ) -> Awaitable[bool]: ...
707 def cluster_meet(
708 self, host: str, port: int, target_nodes: "TargetNodesT" | None = None
709 ) -> bool | Awaitable[bool]:
710 """
711 Force a node cluster to handshake with another node.
712 Sends to specified node.
714 For more information see https://redis.io/commands/cluster-meet
715 """
716 return self.execute_command(
717 "CLUSTER MEET", host, port, target_nodes=target_nodes
718 )
720 @overload
721 def cluster_nodes(self: SyncClientProtocol) -> dict[str, ClusterNodeDetail]: ...
723 @overload
724 def cluster_nodes(
725 self: AsyncClientProtocol,
726 ) -> Awaitable[dict[str, ClusterNodeDetail]]: ...
728 def cluster_nodes(
729 self,
730 ) -> dict[str, ClusterNodeDetail] | Awaitable[dict[str, ClusterNodeDetail]]:
731 """
732 Get Cluster config for the node.
733 Sends to random node in the cluster
735 For more information see https://redis.io/commands/cluster-nodes
736 """
737 return self.execute_command("CLUSTER NODES")
739 @overload
740 def cluster_replicate(
741 self: SyncClientProtocol, target_nodes: "TargetNodesT", node_id: str
742 ) -> bool: ...
744 @overload
745 def cluster_replicate(
746 self: AsyncClientProtocol, target_nodes: "TargetNodesT", node_id: str
747 ) -> Awaitable[bool]: ...
749 def cluster_replicate(
750 self, target_nodes: "TargetNodesT", node_id: str
751 ) -> bool | Awaitable[bool]:
752 """
753 Reconfigure a node as a slave of the specified master node
755 For more information see https://redis.io/commands/cluster-replicate
756 """
757 return self.execute_command(
758 "CLUSTER REPLICATE", node_id, target_nodes=target_nodes
759 )
761 @overload
762 def cluster_reset(
763 self: SyncClientProtocol,
764 soft: bool = True,
765 target_nodes: "TargetNodesT" | None = None,
766 ) -> bool: ...
768 @overload
769 def cluster_reset(
770 self: AsyncClientProtocol,
771 soft: bool = True,
772 target_nodes: "TargetNodesT" | None = None,
773 ) -> Awaitable[bool]: ...
775 def cluster_reset(
776 self, soft: bool = True, target_nodes: "TargetNodesT" | None = None
777 ) -> bool | Awaitable[bool]:
778 """
779 Reset a Redis Cluster node
781 If 'soft' is True then it will send 'SOFT' argument
782 If 'soft' is False then it will send 'HARD' argument
784 For more information see https://redis.io/commands/cluster-reset
785 """
786 return self.execute_command(
787 "CLUSTER RESET", b"SOFT" if soft else b"HARD", target_nodes=target_nodes
788 )
790 @overload
791 def cluster_save_config(
792 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
793 ) -> bool: ...
795 @overload
796 def cluster_save_config(
797 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
798 ) -> Awaitable[bool]: ...
800 def cluster_save_config(
801 self, target_nodes: "TargetNodesT" | None = None
802 ) -> bool | Awaitable[bool]:
803 """
804 Forces the node to save cluster state on disk
806 For more information see https://redis.io/commands/cluster-saveconfig
807 """
808 return self.execute_command("CLUSTER SAVECONFIG", target_nodes=target_nodes)
810 @overload
811 def cluster_get_keys_in_slot(
812 self: SyncClientProtocol, slot: int, num_keys: int
813 ) -> list[bytes | str]: ...
815 @overload
816 def cluster_get_keys_in_slot(
817 self: AsyncClientProtocol, slot: int, num_keys: int
818 ) -> Awaitable[list[bytes | str]]: ...
820 def cluster_get_keys_in_slot(
821 self, slot: int, num_keys: int
822 ) -> list[bytes | str] | Awaitable[list[bytes | str]]:
823 """
824 Returns the number of keys in the specified cluster slot
826 For more information see https://redis.io/commands/cluster-getkeysinslot
827 """
828 return self.execute_command("CLUSTER GETKEYSINSLOT", slot, num_keys)
830 @overload
831 def cluster_set_config_epoch(
832 self: SyncClientProtocol, epoch: int, target_nodes: "TargetNodesT" | None = None
833 ) -> bool: ...
835 @overload
836 def cluster_set_config_epoch(
837 self: AsyncClientProtocol,
838 epoch: int,
839 target_nodes: "TargetNodesT" | None = None,
840 ) -> Awaitable[bool]: ...
842 def cluster_set_config_epoch(
843 self, epoch: int, target_nodes: "TargetNodesT" | None = None
844 ) -> bool | Awaitable[bool]:
845 """
846 Set the configuration epoch in a new node
848 For more information see https://redis.io/commands/cluster-set-config-epoch
849 """
850 return self.execute_command(
851 "CLUSTER SET-CONFIG-EPOCH", epoch, target_nodes=target_nodes
852 )
854 @overload
855 def cluster_setslot(
856 self: SyncClientProtocol,
857 target_node: "TargetNodesT",
858 node_id: str,
859 slot_id: int,
860 state: str,
861 ) -> bool: ...
863 @overload
864 def cluster_setslot(
865 self: AsyncClientProtocol,
866 target_node: "TargetNodesT",
867 node_id: str,
868 slot_id: int,
869 state: str,
870 ) -> Awaitable[bool]: ...
872 def cluster_setslot(
873 self, target_node: "TargetNodesT", node_id: str, slot_id: int, state: str
874 ) -> bool | Awaitable[bool]:
875 """
876 Bind an hash slot to a specific node
878 :target_node: 'ClusterNode'
879 The node to execute the command on
881 For more information see https://redis.io/commands/cluster-setslot
882 """
883 if state.upper() in ("IMPORTING", "NODE", "MIGRATING"):
884 return self.execute_command(
885 "CLUSTER SETSLOT", slot_id, state, node_id, target_nodes=target_node
886 )
887 elif state.upper() == "STABLE":
888 raise RedisError('For "stable" state please use cluster_setslot_stable')
889 else:
890 raise RedisError(f"Invalid slot state: {state}")
892 @overload
893 def cluster_setslot_stable(self: SyncClientProtocol, slot_id: int) -> bool: ...
895 @overload
896 def cluster_setslot_stable(
897 self: AsyncClientProtocol, slot_id: int
898 ) -> Awaitable[bool]: ...
900 def cluster_setslot_stable(self, slot_id: int) -> bool | Awaitable[bool]:
901 """
902 Clears migrating / importing state from the slot.
903 It determines by it self what node the slot is in and sends it there.
905 For more information see https://redis.io/commands/cluster-setslot
906 """
907 return self.execute_command("CLUSTER SETSLOT", slot_id, "STABLE")
909 @overload
910 def cluster_replicas(
911 self: SyncClientProtocol,
912 node_id: str,
913 target_nodes: "TargetNodesT" | None = None,
914 ) -> dict[str, ClusterNodeDetail]: ...
916 @overload
917 def cluster_replicas(
918 self: AsyncClientProtocol,
919 node_id: str,
920 target_nodes: "TargetNodesT" | None = None,
921 ) -> Awaitable[dict[str, ClusterNodeDetail]]: ...
923 def cluster_replicas(
924 self, node_id: str, target_nodes: "TargetNodesT" | None = None
925 ) -> dict[str, ClusterNodeDetail] | Awaitable[dict[str, ClusterNodeDetail]]:
926 """
927 Provides a list of replica nodes replicating from the specified primary
928 target node.
930 For more information see https://redis.io/commands/cluster-replicas
931 """
932 return self.execute_command(
933 "CLUSTER REPLICAS", node_id, target_nodes=target_nodes
934 )
936 @overload
937 def cluster_slots(
938 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
939 ) -> list[Any]: ...
941 @overload
942 def cluster_slots(
943 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
944 ) -> Awaitable[list[Any]]: ...
946 def cluster_slots(
947 self, target_nodes: "TargetNodesT" | None = None
948 ) -> list[Any] | Awaitable[list[Any]]:
949 """
950 Get array of Cluster slot to node mappings
952 For more information see https://redis.io/commands/cluster-slots
953 """
954 return self.execute_command("CLUSTER SLOTS", target_nodes=target_nodes)
956 @overload
957 def cluster_shards(
958 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
959 ) -> ClusterShardsResponse: ...
961 @overload
962 def cluster_shards(
963 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
964 ) -> Awaitable[ClusterShardsResponse]: ...
966 def cluster_shards(
967 self, target_nodes: "TargetNodesT" | None = None
968 ) -> ClusterShardsResponse | Awaitable[ClusterShardsResponse]:
969 """
970 Returns details about the shards of the cluster.
972 For more information see https://redis.io/commands/cluster-shards
973 """
974 return self.execute_command("CLUSTER SHARDS", target_nodes=target_nodes)
976 @overload
977 def cluster_myshardid(
978 self: SyncClientProtocol, target_nodes: "TargetNodesT" | None = None
979 ) -> bytes | str: ...
981 @overload
982 def cluster_myshardid(
983 self: AsyncClientProtocol, target_nodes: "TargetNodesT" | None = None
984 ) -> Awaitable[bytes | str]: ...
986 def cluster_myshardid(self, target_nodes: "TargetNodesT" | None = None) -> (
987 bytes | str
988 ) | Awaitable[bytes | str]:
989 """
990 Returns the shard ID of the node.
992 For more information see https://redis.io/commands/cluster-myshardid/
993 """
994 return self.execute_command("CLUSTER MYSHARDID", target_nodes=target_nodes)
996 @overload
997 def cluster_links(
998 self: SyncClientProtocol, target_node: "TargetNodesT"
999 ) -> ClusterLinksResponse: ...
1001 @overload
1002 def cluster_links(
1003 self: AsyncClientProtocol, target_node: "TargetNodesT"
1004 ) -> Awaitable[ClusterLinksResponse]: ...
1006 def cluster_links(
1007 self, target_node: "TargetNodesT"
1008 ) -> ClusterLinksResponse | Awaitable[ClusterLinksResponse]:
1009 """
1010 Each node in a Redis Cluster maintains a pair of long-lived TCP link with each
1011 peer in the cluster: One for sending outbound messages towards the peer and one
1012 for receiving inbound messages from the peer.
1014 This command outputs information of all such peer links as an array.
1016 For more information see https://redis.io/commands/cluster-links
1017 """
1018 return self.execute_command("CLUSTER LINKS", target_nodes=target_node)
1020 def cluster_flushslots(self, target_nodes: "TargetNodesT" | None = None) -> None:
1021 raise NotImplementedError(
1022 "CLUSTER FLUSHSLOTS is intentionally not implemented in the client."
1023 )
1025 def cluster_bumpepoch(self, target_nodes: "TargetNodesT" | None = None) -> None:
1026 raise NotImplementedError(
1027 "CLUSTER BUMPEPOCH is intentionally not implemented in the client."
1028 )
1030 def readonly(self, target_nodes: "TargetNodesT" | None = None) -> ResponseT:
1031 """
1032 Enables read queries.
1033 The command will be sent to the default cluster node if target_nodes is
1034 not specified.
1036 For more information see https://redis.io/commands/readonly
1037 """
1038 if target_nodes == "replicas" or target_nodes == "all":
1039 # read_from_replicas will only be enabled if the READONLY command
1040 # is sent to all replicas
1041 self.read_from_replicas = True
1042 return self.execute_command("READONLY", target_nodes=target_nodes)
1044 def readwrite(self, target_nodes: "TargetNodesT" | None = None) -> ResponseT:
1045 """
1046 Disables read queries.
1047 The command will be sent to the default cluster node if target_nodes is
1048 not specified.
1050 For more information see https://redis.io/commands/readwrite
1051 """
1052 # Reset read from replicas flag
1053 self.read_from_replicas = False
1054 return self.execute_command("READWRITE", target_nodes=target_nodes)
1056 @deprecated_function(
1057 version="7.2.0",
1058 reason="Use client-side caching feature instead.",
1059 )
1060 def client_tracking_on(
1061 self,
1062 clientid: int | None = None,
1063 prefix: Sequence[KeyT] = [],
1064 bcast: bool = False,
1065 optin: bool = False,
1066 optout: bool = False,
1067 noloop: bool = False,
1068 target_nodes: "TargetNodesT" | None = "all",
1069 ) -> ResponseT:
1070 """
1071 Enables the tracking feature of the Redis server, that is used
1072 for server assisted client side caching.
1074 When clientid is provided - in target_nodes only the node that owns the
1075 connection with this id should be provided.
1076 When clientid is not provided - target_nodes can be any node.
1078 For more information see https://redis.io/commands/client-tracking
1079 """
1080 return self.client_tracking(
1081 True,
1082 clientid,
1083 prefix,
1084 bcast,
1085 optin,
1086 optout,
1087 noloop,
1088 target_nodes=target_nodes,
1089 )
1091 @deprecated_function(
1092 version="7.2.0",
1093 reason="Use client-side caching feature instead.",
1094 )
1095 def client_tracking_off(
1096 self,
1097 clientid: int | None = None,
1098 prefix: Sequence[KeyT] = [],
1099 bcast: bool = False,
1100 optin: bool = False,
1101 optout: bool = False,
1102 noloop: bool = False,
1103 target_nodes: "TargetNodesT" | None = "all",
1104 ) -> ResponseT:
1105 """
1106 Disables the tracking feature of the Redis server, that is used
1107 for server assisted client side caching.
1109 When clientid is provided - in target_nodes only the node that owns the
1110 connection with this id should be provided.
1111 When clientid is not provided - target_nodes can be any node.
1113 For more information see https://redis.io/commands/client-tracking
1114 """
1115 return self.client_tracking(
1116 False,
1117 clientid,
1118 prefix,
1119 bcast,
1120 optin,
1121 optout,
1122 noloop,
1123 target_nodes=target_nodes,
1124 )
1126 def hotkeys_start(
1127 self,
1128 metrics: List[HotkeysMetricsTypes],
1129 count: int | None = None,
1130 duration: int | None = None,
1131 sample_ratio: int | None = None,
1132 slots: List[int] | None = None,
1133 **kwargs,
1134 ) -> str | bytes:
1135 """
1136 Cluster client does not support hotkeys command. Please use the non-cluster client.
1138 For more information see https://redis.io/commands/hotkeys-start
1139 """
1140 raise NotImplementedError(
1141 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1142 )
1144 def hotkeys_stop(self, **kwargs) -> str | bytes:
1145 """
1146 Cluster client does not support hotkeys command. Please use the non-cluster client.
1148 For more information see https://redis.io/commands/hotkeys-stop
1149 """
1150 raise NotImplementedError(
1151 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1152 )
1154 def hotkeys_reset(self, **kwargs) -> str | bytes:
1155 """
1156 Cluster client does not support hotkeys command. Please use the non-cluster client.
1158 For more information see https://redis.io/commands/hotkeys-reset
1159 """
1160 raise NotImplementedError(
1161 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1162 )
1164 def hotkeys_get(self, **kwargs) -> list[dict[str | bytes, Any]]:
1165 """
1166 Cluster client does not support hotkeys command. Please use the non-cluster client.
1168 For more information see https://redis.io/commands/hotkeys-get
1169 """
1170 raise NotImplementedError(
1171 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1172 )
1175class AsyncClusterManagementCommands(
1176 ClusterManagementCommands, AsyncManagementCommands
1177):
1178 """
1179 A class for Redis Cluster management commands
1181 The class inherits from Redis's core ManagementCommands class and do the
1182 required adjustments to work with cluster mode
1183 """
1185 async def cluster_delslots(self, *slots: EncodableT) -> List[bool]:
1186 """
1187 Set hash slots as unbound in the cluster.
1188 It determines by it self what node the slot is in and sends it there
1190 Returns a list of the results for each processed slot.
1192 For more information see https://redis.io/commands/cluster-delslots
1193 """
1194 return await asyncio.gather(
1195 *(
1196 asyncio.create_task(self.execute_command("CLUSTER DELSLOTS", slot))
1197 for slot in slots
1198 )
1199 )
1201 @deprecated_function(
1202 version="7.2.0",
1203 reason="Use client-side caching feature instead.",
1204 )
1205 async def client_tracking_on(
1206 self,
1207 clientid: int | None = None,
1208 prefix: Sequence[KeyT] = [],
1209 bcast: bool = False,
1210 optin: bool = False,
1211 optout: bool = False,
1212 noloop: bool = False,
1213 target_nodes: "TargetNodesT" | None = "all",
1214 ) -> ResponseT:
1215 """
1216 Enables the tracking feature of the Redis server, that is used
1217 for server assisted client side caching.
1219 When clientid is provided - in target_nodes only the node that owns the
1220 connection with this id should be provided.
1221 When clientid is not provided - target_nodes can be any node.
1223 For more information see https://redis.io/commands/client-tracking
1224 """
1225 return await self.client_tracking(
1226 True,
1227 clientid,
1228 prefix,
1229 bcast,
1230 optin,
1231 optout,
1232 noloop,
1233 target_nodes=target_nodes,
1234 )
1236 @deprecated_function(
1237 version="7.2.0",
1238 reason="Use client-side caching feature instead.",
1239 )
1240 async def client_tracking_off(
1241 self,
1242 clientid: int | None = None,
1243 prefix: Sequence[KeyT] = [],
1244 bcast: bool = False,
1245 optin: bool = False,
1246 optout: bool = False,
1247 noloop: bool = False,
1248 target_nodes: "TargetNodesT" | None = "all",
1249 ) -> ResponseT:
1250 """
1251 Disables the tracking feature of the Redis server, that is used
1252 for server assisted client side caching.
1254 When clientid is provided - in target_nodes only the node that owns the
1255 connection with this id should be provided.
1256 When clientid is not provided - target_nodes can be any node.
1258 For more information see https://redis.io/commands/client-tracking
1259 """
1260 return await self.client_tracking(
1261 False,
1262 clientid,
1263 prefix,
1264 bcast,
1265 optin,
1266 optout,
1267 noloop,
1268 target_nodes=target_nodes,
1269 )
1271 async def hotkeys_start(
1272 self,
1273 metrics: List[HotkeysMetricsTypes],
1274 count: int | None = None,
1275 duration: int | None = None,
1276 sample_ratio: int | None = None,
1277 slots: List[int] | None = None,
1278 **kwargs,
1279 ) -> Awaitable[str | bytes]:
1280 """
1281 Cluster client does not support hotkeys command. Please use the non-cluster client.
1283 For more information see https://redis.io/commands/hotkeys-start
1284 """
1285 raise NotImplementedError(
1286 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1287 )
1289 async def hotkeys_stop(self, **kwargs) -> Awaitable[str | bytes]:
1290 """
1291 Cluster client does not support hotkeys command. Please use the non-cluster client.
1293 For more information see https://redis.io/commands/hotkeys-stop
1294 """
1295 raise NotImplementedError(
1296 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1297 )
1299 async def hotkeys_reset(self, **kwargs) -> Awaitable[str | bytes]:
1300 """
1301 Cluster client does not support hotkeys command. Please use the non-cluster client.
1303 For more information see https://redis.io/commands/hotkeys-reset
1304 """
1305 raise NotImplementedError(
1306 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1307 )
1309 async def hotkeys_get(self, **kwargs) -> Awaitable[list[dict[str | bytes, Any]]]:
1310 """
1311 Cluster client does not support hotkeys command. Please use the non-cluster client.
1313 For more information see https://redis.io/commands/hotkeys-get
1314 """
1315 raise NotImplementedError(
1316 "HOTKEYS commands are not supported in cluster mode. Please use the non-cluster client."
1317 )
1320class ClusterDataAccessCommands(DataAccessCommands):
1321 """
1322 A class for Redis Cluster Data Access Commands
1324 The class inherits from Redis's core DataAccessCommand class and do the
1325 required adjustments to work with cluster mode
1326 """
1328 @overload
1329 def stralgo(
1330 self: SyncClientProtocol,
1331 algo: Literal["LCS"],
1332 value1: KeyT,
1333 value2: KeyT,
1334 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1335 len: bool = False,
1336 idx: bool = False,
1337 minmatchlen: int | None = None,
1338 withmatchlen: bool = False,
1339 **kwargs,
1340 ) -> StralgoResponse: ...
1342 @overload
1343 def stralgo(
1344 self: AsyncClientProtocol,
1345 algo: Literal["LCS"],
1346 value1: KeyT,
1347 value2: KeyT,
1348 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1349 len: bool = False,
1350 idx: bool = False,
1351 minmatchlen: int | None = None,
1352 withmatchlen: bool = False,
1353 **kwargs,
1354 ) -> Awaitable[StralgoResponse]: ...
1356 def stralgo(
1357 self,
1358 algo: Literal["LCS"],
1359 value1: KeyT,
1360 value2: KeyT,
1361 specific_argument: Literal["strings"] | Literal["keys"] = "strings",
1362 len: bool = False,
1363 idx: bool = False,
1364 minmatchlen: int | None = None,
1365 withmatchlen: bool = False,
1366 **kwargs,
1367 ) -> StralgoResponse | Awaitable[StralgoResponse]:
1368 """
1369 Implements complex algorithms that operate on strings.
1370 Right now the only algorithm implemented is the LCS algorithm
1371 (longest common substring). However new algorithms could be
1372 implemented in the future.
1374 ``algo`` Right now must be LCS
1375 ``value1`` and ``value2`` Can be two strings or two keys
1376 ``specific_argument`` Specifying if the arguments to the algorithm
1377 will be keys or strings. strings is the default.
1378 ``len`` Returns just the len of the match.
1379 ``idx`` Returns the match positions in each string.
1380 ``minmatchlen`` Restrict the list of matches to the ones of a given
1381 minimal length. Can be provided only when ``idx`` set to True.
1382 ``withmatchlen`` Returns the matches with the len of the match.
1383 Can be provided only when ``idx`` set to True.
1385 For more information see https://redis.io/commands/stralgo
1386 """
1387 target_nodes = kwargs.pop("target_nodes", None)
1388 if specific_argument == "strings" and target_nodes is None:
1389 target_nodes = "default-node"
1390 kwargs.update({"target_nodes": target_nodes})
1391 return super().stralgo(
1392 algo,
1393 value1,
1394 value2,
1395 specific_argument,
1396 len,
1397 idx,
1398 minmatchlen,
1399 withmatchlen,
1400 **kwargs,
1401 )
1403 def scan_iter(
1404 self,
1405 match: PatternT | None = None,
1406 count: int | None = None,
1407 _type: str | None = None,
1408 **kwargs,
1409 ) -> Iterator:
1410 # Do the first query with cursor=0 for all nodes
1411 cursors, data = self.scan(match=match, count=count, _type=_type, **kwargs)
1412 yield from data
1414 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1415 if cursors:
1416 # Get nodes by name
1417 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1419 # Iterate over each node till its cursor is 0
1420 kwargs.pop("target_nodes", None)
1421 while cursors:
1422 for name, cursor in cursors.items():
1423 cur, data = self.scan(
1424 cursor=cursor,
1425 match=match,
1426 count=count,
1427 _type=_type,
1428 target_nodes=nodes[name],
1429 **kwargs,
1430 )
1431 yield from data
1432 cursors[name] = cur[name]
1434 cursors = {
1435 name: cursor for name, cursor in cursors.items() if cursor != 0
1436 }
1439class AsyncClusterDataAccessCommands(
1440 ClusterDataAccessCommands, AsyncDataAccessCommands
1441):
1442 """
1443 A class for Redis Cluster Data Access Commands
1445 The class inherits from Redis's core DataAccessCommand class and do the
1446 required adjustments to work with cluster mode
1447 """
1449 async def scan_iter(
1450 self,
1451 match: PatternT | None = None,
1452 count: int | None = None,
1453 _type: str | None = None,
1454 **kwargs,
1455 ) -> AsyncIterator:
1456 # Do the first query with cursor=0 for all nodes
1457 cursors, data = await self.scan(match=match, count=count, _type=_type, **kwargs)
1458 for value in data:
1459 yield value
1461 cursors = {name: cursor for name, cursor in cursors.items() if cursor != 0}
1462 if cursors:
1463 # Get nodes by name
1464 nodes = {name: self.get_node(node_name=name) for name in cursors.keys()}
1466 # Iterate over each node till its cursor is 0
1467 kwargs.pop("target_nodes", None)
1468 while cursors:
1469 for name, cursor in cursors.items():
1470 cur, data = await self.scan(
1471 cursor=cursor,
1472 match=match,
1473 count=count,
1474 _type=_type,
1475 target_nodes=nodes[name],
1476 **kwargs,
1477 )
1478 for value in data:
1479 yield value
1480 cursors[name] = cur[name]
1482 cursors = {
1483 name: cursor for name, cursor in cursors.items() if cursor != 0
1484 }
1487class RedisClusterCommands(
1488 ClusterMultiKeyCommands,
1489 ClusterManagementCommands,
1490 ACLCommands,
1491 PubSubCommands,
1492 ClusterDataAccessCommands,
1493 ScriptCommands,
1494 FunctionCommands,
1495 ModuleCommands,
1496 RedisModuleCommands,
1497):
1498 """
1499 A class for all Redis Cluster commands
1501 For key-based commands, the target node(s) will be internally determined
1502 by the keys' hash slot.
1503 Non-key-based commands can be executed with the 'target_nodes' argument to
1504 target specific nodes. By default, if target_nodes is not specified, the
1505 command will be executed on the default cluster node.
1507 :param :target_nodes: type can be one of the followings:
1508 - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
1509 - 'ClusterNode'
1510 - 'list(ClusterNodes)'
1511 - 'dict(any:clusterNodes)'
1513 for example:
1514 r.cluster_info(target_nodes=RedisCluster.ALL_NODES)
1515 """
1518class AsyncRedisClusterCommands(
1519 AsyncClusterMultiKeyCommands,
1520 AsyncClusterManagementCommands,
1521 AsyncACLCommands,
1522 PubSubCommands,
1523 AsyncClusterDataAccessCommands,
1524 AsyncScriptCommands,
1525 AsyncFunctionCommands,
1526 AsyncModuleCommands,
1527 AsyncRedisModuleCommands,
1528):
1529 """
1530 A class for all Redis Cluster commands
1532 For key-based commands, the target node(s) will be internally determined
1533 by the keys' hash slot.
1534 Non-key-based commands can be executed with the 'target_nodes' argument to
1535 target specific nodes. By default, if target_nodes is not specified, the
1536 command will be executed on the default cluster node.
1538 :param :target_nodes: type can be one of the followings:
1539 - nodes flag: ALL_NODES, PRIMARIES, REPLICAS, RANDOM
1540 - 'ClusterNode'
1541 - 'list(ClusterNodes)'
1542 - 'dict(any:clusterNodes)'
1544 for example:
1545 r.cluster_info(target_nodes=RedisCluster.ALL_NODES)
1546 """