def _suggest_cluster_id()

in aws_advanced_python_wrapper/host_list_provider.py [0:0]

• 15 lines of code
• 9 McCabe index (conditional complexity)

    def _suggest_cluster_id(self, primary_cluster_id_hosts: Tuple[HostInfo, ...]):
        if not primary_cluster_id_hosts:
            return

        primary_cluster_id_urls = {host.url for host in primary_cluster_id_hosts}
        for cluster_id, hosts in RdsHostListProvider._topology_cache.get_dict().items():
            is_primary_cluster = RdsHostListProvider._is_primary_cluster_id_cache.get_with_default(
                cluster_id, False, self._suggested_cluster_id_refresh_ns)
            suggested_primary_cluster_id = RdsHostListProvider._cluster_ids_to_update.get(cluster_id)
            if is_primary_cluster or suggested_primary_cluster_id or not hosts:
                continue

            # The entry is non-primary
            for host in hosts:
                if host.url in primary_cluster_id_urls:
                    # An instance URL in this topology cache entry matches an instance URL in the primary cluster entry.
                    # The associated cluster ID should be updated to match the primary ID so that they can share
                    # topology info.
                    RdsHostListProvider._cluster_ids_to_update.put(
                        cluster_id, self._cluster_id, self._suggested_cluster_id_refresh_ns)
                    break