def merge_from_dict()

in common/py_libs/config_spec.py [0:0]

• 53 lines of code
• 16 McCabe index (conditional complexity)

    def merge_from_dict(cls: Type[T], a_dict: Dict[str, Any], b_dict: Dict[str,
                                                                           Any],
                        type_hints: Dict[str, Any]) -> Dict[str, Any]:
        """Returns a merged ConfigSpec represented as dictionaries.

        Both inputs and output are ConfigSpecs represented as dictionaries.
        Values from `a` will take precedence over `b` where they differ and are
        valid. For Lists, if the elements have a common `name` or `display_name`
        they will be considered the same element and will be merged.
        """
        for key, a_value in a_dict.items():
            if key not in b_dict:
                b_dict[key] = a_value
                continue

            # Ignore invalid values.
            if not a_value:
                continue

            field_type = type_hints[key]

            # For single values.
            if not isinstance(a_value, list):
                if issubclass(field_type, ConfigSpec):
                    item_type_hints = typing.get_type_hints(field_type)
                    nested_configspec = field_type.merge_from_dict(
                        a_value, b_dict[key], item_type_hints)
                    b_dict[key] = nested_configspec
                else:
                    b_dict[key] = a_value
                continue

            # Get element field types for lists.
            element_field_types = _unwrap_field_type(field_type)
            if len(element_field_types) == 0:
                raise cortex_exc.TypeCError(
                    f"Unable to get nested types for {field_type}")
            if len(element_field_types) > 1:
                raise cortex_exc.NotImplementedCError(
                    "ConfigSpec merge with union field types isn't supported.")
            element_field_type = element_field_types[0]

            # For valid non-ConfigSpec lists, a overwrites b.
            if not issubclass(element_field_type, ConfigSpec):
                b_dict[key] = a_value
                continue

            # For repeated ConfigSpecs:
            if issubclass(element_field_type, ConfigSpec):
                element_type_hints = typing.get_type_hints(element_field_type)

                # If the ConfigSpec doesn't have a name field, behaves the same
                # as non-ConfigSpec lists, where a just overwrites b.
                try:
                    _ = _get_name(element_type_hints)
                except cortex_exc.KeyCError:
                    b_dict[key] = a_value
                    continue

                # For named repeated ConfigSpecs, merge elements with same
                # name/display_name.
                b_element_map = {_get_name(e): e for e in b_dict[key]}
                for a_element in a_value:
                    name = _get_name(a_element)
                    if name in b_element_map:
                        merged_element = element_field_type.merge_from_dict(
                            a_element, b_element_map[name], element_type_hints)
                        b_element_map[name] = merged_element
                    # Add elements with new name/display_name.
                    else:
                        b_element_map[name] = a_element

                # Make sure the final list is ordered to be deterministic.
                ordered_elements = [v for k, v in sorted(b_element_map.items())]
                b_dict[key] = ordered_elements
                continue

            raise cortex_exc.TypeCError(
                f"Unexpected field type '{a_value.__class__.__name__}' while "
                f"merging instances of '{cls.__name__}'")

        return b_dict