def _compare_section()

in cli/src/pcluster/config/config_patch.py [0:0]

• 36 lines of code
• 14 McCabe index (conditional complexity)

    def _compare_section(self, base_section: dict, target_section: dict, section_schema: BaseSchema, param_path: List):
        """
        Compare the provided base and target sections and append the detected changes to the internal changes list.

        :param base_section: The section in the base configuration
        :param target_section: The corresponding section in the target configuration
        :param section_schema: schema corresponding to the section to be analyzed (contains all the resources/params)
        :param param_path: A list on which the items correspond to the path of the param in the configuration schema
        """
        for _, field_obj in section_schema.declared_fields.items():
            data_key = field_obj.data_key
            is_nested_section = hasattr(field_obj, "nested")
            is_list = hasattr(field_obj, "many") and field_obj.many

            change_update_policy = field_obj.metadata.get("update_policy", UpdatePolicy.UNSUPPORTED)

            if is_nested_section:
                if is_list:
                    self._compare_list(
                        base_section, target_section, param_path, data_key, field_obj, change_update_policy
                    )
                else:
                    # Single nested section
                    target_value = target_section.get(data_key, None)
                    base_value = base_section.get(data_key, None)

                    if target_value and base_value:
                        # Compare nested sections and params
                        nested_path = copy.deepcopy(param_path)
                        nested_path.append(data_key)
                        self._compare_section(base_value, target_value, field_obj.schema, nested_path)
                    elif target_value or base_value:
                        # One section has been added or removed, add section change information
                        self.changes.append(
                            Change(
                                param_path,
                                data_key,
                                base_value if base_value else "-",
                                target_value if target_value else "-",
                                change_update_policy,
                                is_list=False,
                            )
                        )
            else:
                # Simple param
                target_value = target_section.get(data_key, None)
                base_value = base_section.get(data_key, None)

                if target_value != base_value:
                    # Add param change information
                    self.changes.append(
                        Change(param_path, data_key, base_value, target_value, change_update_policy, is_list=False)
                    )