def _determine_nodes()

in redis/cluster.py [0:0]

• 29 lines of code
• 13 McCabe index (conditional complexity)

    def _determine_nodes(self, *args, **kwargs) -> List["ClusterNode"]:
        # Determine which nodes should be executed the command on.
        # Returns a list of target nodes.
        command = args[0].upper()
        if len(args) >= 2 and f"{args[0]} {args[1]}".upper() in self.command_flags:
            command = f"{args[0]} {args[1]}".upper()

        nodes_flag = kwargs.pop("nodes_flag", None)
        if nodes_flag is not None:
            # nodes flag passed by the user
            command_flag = nodes_flag
        else:
            # get the nodes group for this command if it was predefined
            command_flag = self.command_flags.get(command)
        if command_flag == self.__class__.RANDOM:
            # return a random node
            return [self.get_random_node()]
        elif command_flag == self.__class__.PRIMARIES:
            # return all primaries
            return self.get_primaries()
        elif command_flag == self.__class__.REPLICAS:
            # return all replicas
            return self.get_replicas()
        elif command_flag == self.__class__.ALL_NODES:
            # return all nodes
            return self.get_nodes()
        elif command_flag == self.__class__.DEFAULT_NODE:
            # return the cluster's default node
            return [self.nodes_manager.default_node]
        elif command in self.__class__.SEARCH_COMMANDS[0]:
            return [self.nodes_manager.default_node]
        else:
            # get the node that holds the key's slot
            slot = self.determine_slot(*args)
            node = self.nodes_manager.get_node_from_slot(
                slot,
                self.read_from_replicas and command in READ_COMMANDS,
                self.load_balancing_strategy if command in READ_COMMANDS else None,
            )
            return [node]