(
self, *args, request_policy: RequestPolicy, **kwargs
)
| 4325 | return nodes |
| 4326 | |
| 4327 | def _determine_nodes( |
| 4328 | self, *args, request_policy: RequestPolicy, **kwargs |
| 4329 | ) -> List["ClusterNode"]: |
| 4330 | # Determine which nodes should be executed the command on. |
| 4331 | # Returns a list of target nodes. |
| 4332 | command = args[0].upper() |
| 4333 | if ( |
| 4334 | len(args) >= 2 |
| 4335 | and f"{args[0]} {args[1]}".upper() in self._pipe.command_flags |
| 4336 | ): |
| 4337 | command = f"{args[0]} {args[1]}".upper() |
| 4338 | |
| 4339 | nodes_flag = kwargs.pop("nodes_flag", None) |
| 4340 | if nodes_flag is not None: |
| 4341 | # nodes flag passed by the user |
| 4342 | command_flag = nodes_flag |
| 4343 | else: |
| 4344 | # get the nodes group for this command if it was predefined |
| 4345 | command_flag = self._pipe.command_flags.get(command) |
| 4346 | |
| 4347 | if command_flag in self._pipe._command_flags_mapping: |
| 4348 | request_policy = self._pipe._command_flags_mapping[command_flag] |
| 4349 | |
| 4350 | policy_callback = self._pipe._policies_callback_mapping[request_policy] |
| 4351 | |
| 4352 | if request_policy == RequestPolicy.DEFAULT_KEYED: |
| 4353 | nodes = policy_callback(command, *args) |
| 4354 | elif request_policy == RequestPolicy.MULTI_SHARD: |
| 4355 | nodes = policy_callback(*args, **kwargs) |
| 4356 | elif request_policy == RequestPolicy.DEFAULT_KEYLESS: |
| 4357 | nodes = policy_callback(args[0]) |
| 4358 | else: |
| 4359 | nodes = policy_callback() |
| 4360 | |
| 4361 | if args[0].lower() == "ft.aggregate": |
| 4362 | self._aggregate_nodes = nodes |
| 4363 | |
| 4364 | return nodes |
| 4365 | |
| 4366 | def multi(self): |
| 4367 | raise RedisClusterException( |
no test coverage detected