Method _execute

redis/asyncio/cluster.py:2613–2752

(
        self,
        client: "RedisCluster",
        stack: List["PipelineCommand"],
        raise_on_error: bool = True,
        allow_redirections: bool = True,
    )

            await self.reset()

    async def _execute(
        self,
        client: "RedisCluster",
        stack: List["PipelineCommand"],
        raise_on_error: bool = True,
        allow_redirections: bool = True,
    ) -> List[Any]:
        todo = [
            cmd for cmd in stack if not cmd.result or isinstance(cmd.result, Exception)
        ]

        nodes = {}
        for cmd in todo:
            passed_targets = cmd.kwargs.pop("target_nodes", None)
            command_policies = await client._policy_resolver.resolve(
                cmd.args[0].lower()
            )

            if passed_targets and not client._is_node_flag(passed_targets):
                target_nodes = client._parse_target_nodes(passed_targets)

                if not command_policies:
                    command_policies = CommandPolicies()
            else:
                if not command_policies:
                    command_flag = client.command_flags.get(cmd.args[0])
                    if not command_flag:
                        # Fallback to default policy
                        if not client.get_default_node():
                            slot = None
                        else:
                            slot = await client._determine_slot(*cmd.args)
                        if slot is None:
                            command_policies = CommandPolicies()
                        else:
                            command_policies = CommandPolicies(
                                request_policy=RequestPolicy.DEFAULT_KEYED,
                                response_policy=ResponsePolicy.DEFAULT_KEYED,
                            )
                    else:
                        if command_flag in client._command_flags_mapping:
                            command_policies = CommandPolicies(
                                request_policy=client._command_flags_mapping[
                                    command_flag
                                ]
                            )
                        else:
                            command_policies = CommandPolicies()

                target_nodes = await client._determine_nodes(
                    *cmd.args,
                    request_policy=command_policies.request_policy,
                    node_flag=passed_targets,
                )
                if not target_nodes:
                    raise RedisClusterException(
                        f"No targets were found to execute {cmd.args} command on"
                    )