def get_valid_instance_types_by_allowed_list()

in source/idea/idea-virtual-desktop-controller/src/ideavirtualdesktopcontroller/app/virtual_desktop_controller_utils.py [0:0]

• 47 lines of code
• 18 McCabe index (conditional complexity)

    def get_valid_instance_types_by_allowed_list(self, hibernation_support: bool, allowed_instance_types: List[str]) -> Dict:
        instance_types_names = self.context.cache().long_term().get(self.INSTANCE_TYPES_NAMES_LIST_CACHE_KEY)
        instance_info_data = self.context.cache().long_term().get(self.INSTANCE_INFO_CACHE_KEY)
        if instance_types_names is None or instance_info_data is None:
            # not found in cache, need to update it again.
            self._add_instance_data_to_cache()
            instance_types_names = self.context.cache().long_term().get(self.INSTANCE_TYPES_NAMES_LIST_CACHE_KEY)
            instance_info_data = self.context.cache().long_term().get(self.INSTANCE_INFO_CACHE_KEY)

        # We now have a list of all instance types (Cache has been updated IF it was empty).
        valid_instance_types_dict = {}

        allowed_instance_type_names = set()
        allowed_instance_type_families = set()
        for instance_type in allowed_instance_types:
            if '.' in instance_type:
                allowed_instance_type_names.add(instance_type)
            else:
                allowed_instance_type_families.add(instance_type)

        denied_instance_types = set(self.context.config().get_list('virtual-desktop-controller.dcv_session.instance_types.deny', default=[]))
        denied_instance_type_names = set()
        denied_instance_type_families = set()
        for instance_type in denied_instance_types:
            if '.' in instance_type:
                denied_instance_type_names.add(instance_type)
            else:
                denied_instance_type_families.add(instance_type)

        if self._logger.isEnabledFor(logging.DEBUG):
            self._logger.debug(f"get_valid_instance_types() - Instance Allow/Deny Summary")
            self._logger.debug(f"Allowed instance Families: {allowed_instance_type_families}")
            self._logger.debug(f"Allowed instances: {allowed_instance_type_names}")
            self._logger.debug(f"Denied instance Families: {denied_instance_type_families}")
            self._logger.debug(f"Denied instances: {denied_instance_type_names}")

        for instance_type_name in instance_types_names:
            instance_type_family = instance_type_name.split('.')[0]
            self._logger.debug(f"Processing - Instance Name: {instance_type_name}   Family: {instance_type_family}")

            if instance_type_name not in allowed_instance_type_names and instance_type_family not in allowed_instance_type_families:
                # instance type or instance family is not present in allow list
                self._logger.debug(f"Found {instance_type_name} ({instance_type_family}) NOT in ALLOW config: ({allowed_instance_type_names} / {allowed_instance_type_families})")
                continue

            if instance_type_name in denied_instance_type_names or instance_type_family in denied_instance_type_families:
                # instance type or instance family is present in deny list
                self._logger.debug(f"Found {instance_type_name} ({instance_type_family}) IN DENIED config: ({denied_instance_type_names} / {denied_instance_type_families})")
                continue

            instance_info = instance_info_data[instance_type_name]
            hibernation_supported = Utils.get_value_as_bool('HibernationSupported', instance_info, default=False)
            if hibernation_support and not hibernation_supported:
                self._logger.debug(f"Hibernation ({hibernation_support}) != Instance {instance_type_name} ({hibernation_supported}) Skipped.")
                continue
            
            # All checks passed if we make it this far
            self._logger.debug(f"Instance {instance_type_name} - Added as valid_instance_types")
            valid_instance_types_dict[instance_type_name] = instance_info
        self._logger.debug(f"Returning valid_instance_types: {valid_instance_types_dict.keys()}")
        return valid_instance_types_dict