def get_host()

in aws_advanced_python_wrapper/host_selector.py [0:0]

• 28 lines of code
• 17 McCabe index (conditional complexity)

    def get_host(self, hosts: Tuple[HostInfo, ...], role: HostRole, props: Optional[Properties] = None) -> HostInfo:

        eligible_hosts: List[HostInfo] = [host for host in hosts if host.role == role and host.get_availability() == HostAvailability.AVAILABLE]
        eligible_hosts.sort(key=lambda host: host.host, reverse=False)
        if len(eligible_hosts) == 0:
            raise AwsWrapperError(Messages.get_formatted("HostSelector.NoHostsMatchingRole", role))

        # Create new cache entries for provided hosts if necessary. All hosts point to the same cluster info.
        self._create_cache_entry_for_hosts(eligible_hosts, props)
        current_cluster_info_key: str = eligible_hosts[0].host
        cluster_info: Optional[RoundRobinClusterInfo] = RoundRobinHostSelector._round_robin_cache.get(current_cluster_info_key)

        last_host_index: int = -1
        if cluster_info is None:
            raise AwsWrapperError(Messages.get("RoundRobinHostSelector.ClusterInfoNone"))

        last_host = cluster_info.last_host
        # Check if last_host is in list of eligible hosts. Update last_host_index.
        if last_host is not None:
            for i in range(0, len(eligible_hosts)):
                if eligible_hosts[i].host == last_host.host:
                    last_host_index = i

        if cluster_info.weight_counter > 0 and last_host_index != -1:
            target_host_index = last_host_index
        else:
            if last_host_index != -1 and last_host_index != (len(eligible_hosts) - 1):
                target_host_index = last_host_index + 1
            else:
                target_host_index = 0
            weight = cluster_info.cluster_weights_dict.get(eligible_hosts[target_host_index].host)
            cluster_info.weight_counter = cluster_info.default_weight if weight is None else weight

        cluster_info.weight_counter = (cluster_info.weight_counter - 1)
        cluster_info.last_host = eligible_hosts[target_host_index]
        return eligible_hosts[target_host_index]