Method _execute_command

redis/asyncio/cluster.py:1098–1283

(
        self, target_node: "ClusterNode", *args: Union[KeyT, EncodableT], **kwargs: Any
    )

                    raise e

    async def _execute_command(
        self, target_node: "ClusterNode", *args: Union[KeyT, EncodableT], **kwargs: Any
    ) -> Any:
        asking = moved = False
        redirect_addr = None
        ttl = self.RedisClusterRequestTTL
        command = args[0]
        start_time = time.monotonic()

        while ttl > 0:
            ttl -= 1
            try:
                if asking:
                    target_node = self.get_node(node_name=redirect_addr)
                    await target_node.execute_command("ASKING")
                    asking = False
                elif moved:
                    # MOVED occurred and the slots cache was updated,
                    # refresh the target node
                    slot = await self._determine_slot(*args)
                    target_node = self.nodes_manager.get_node_from_slot(
                        slot,
                        self.read_from_replicas and args[0] in READ_COMMANDS,
                        self.load_balancing_strategy
                        if args[0] in READ_COMMANDS
                        else None,
                    )
                    moved = False

                response = await target_node.execute_command(*args, **kwargs)
                await self._record_command_metric(
                    command_name=command,
                    duration_seconds=time.monotonic() - start_time,
                    connection=target_node,
                )
                return response
            except BusyLoadingError as e:
                e.connection = target_node
                await self._record_command_metric(
                    command_name=command,
                    duration_seconds=time.monotonic() - start_time,
                    connection=target_node,
                    error=e,
                )
                raise
            except MaxConnectionsError as e:
                # MaxConnectionsError indicates client-side resource exhaustion
                # (too many connections in the pool), not a node failure.
                # Don't treat this as a node failure - just re-raise the error
                # without reinitializing the cluster.
                e.connection = target_node
                await self._record_command_metric(
                    command_name=command,
                    duration_seconds=time.monotonic() - start_time,
                    connection=target_node,
                    error=e,
                )
                raise