src/buildstream/_includes.py

import os
from . import _yaml
from .node import MappingNode, ScalarNode, SequenceNode
from ._exceptions import LoadError
from .exceptions import LoadErrorReason


# Includes()
#
# This takes care of processing include directives "(@)".
#
# Args:
#    loader (Loader): The Loader object
#    copy_tree (bool): Whether to make a copy, of tree in
#                      provenance. Should be true if intended to be
#                      serialized.
class Includes:
    def __init__(self, loader, *, copy_tree=False):
        self._loader = loader
        self._loaded = {}
        self._copy_tree = copy_tree

    # process()
    #
    # Process recursively include directives in a YAML node.
    #
    # Args:
    #    node (dict): A YAML node
    #    only_local (bool): Whether to ignore junction files
    #    first_pass (bool): Whether to use first pass options.
    #    process_options (bool): Whether to process options.
    def process(self, node, *, only_local=False, first_pass=False, process_options=True):
        self._process(node, only_local=only_local, first_pass=first_pass, process_options=process_options)

    # _process()
    #
    # Process recursively include directives in a YAML node. This
    # method is a recursively called on loaded nodes from files.
    #
    # Args:
    #    node (dict): A YAML node
    #    included (set): Fail for recursion if trying to load any files in this set
    #    current_loader (Loader): Use alternative loader (for junction files)
    #    only_local (bool): Whether to ignore junction files
    #    first_pass (bool): Whether to use first pass options.
    #    process_options (bool): Whether to process options.
    def _process(
        self, node, *, included=None, current_loader=None, only_local=False, first_pass=False, process_options=True,
    ):
        if current_loader is None:
            current_loader = self._loader

        if not process_options:
            pass
        elif first_pass:
            current_loader.project.first_pass_config.options.process_node(node)
        else:
            current_loader.project.ensure_fully_loaded()
            current_loader.project.options.process_node(node)

        self._process_node(
            node,
            included=included,
            only_local=only_local,
            current_loader=current_loader,
            first_pass=first_pass,
            process_options=process_options,
        )

    # _process_node()
    #
    # Process recursively include directives in a YAML node. This
    # method is recursively called on all nodes.
    #
    # Args:
    #    node (dict): A YAML node
    #    included (set): Fail for recursion if trying to load any files in this set
    #    current_loader (Loader): Use alternative loader (for junction files)
    #    only_local (bool): Whether to ignore junction files
    #    first_pass (bool): Whether to use first pass options.
    #    process_options (bool): Whether to process options.
    def _process_node(
        self, node, *, included=None, current_loader=None, only_local=False, first_pass=False, process_options=True,
    ):
        if included is None:
            included = set()

        includes_node = node.get_node("(@)", allowed_types=[ScalarNode, SequenceNode], allow_none=True)

        if includes_node:
            if type(includes_node) is ScalarNode:  # pylint: disable=unidiomatic-typecheck
                includes = [includes_node.as_str()]
            else:
                includes = includes_node.as_str_list()

            del node["(@)"]

            for include in reversed(includes):
                if only_local and ":" in include:
                    continue
                try:
                    include_node, file_path, sub_loader = self._include_file(include, current_loader)
                except LoadError as e:
                    include_provenance = includes_node.get_provenance()
                    if e.reason == LoadErrorReason.MISSING_FILE:
                        message = "{}: Include block references a file that could not be found: '{}'.".format(
                            include_provenance, include
                        )
                        raise LoadError(message, LoadErrorReason.MISSING_FILE) from e
                    if e.reason == LoadErrorReason.LOADING_DIRECTORY:
                        message = "{}: Include block references a directory instead of a file: '{}'.".format(
                            include_provenance, include
                        )
                        raise LoadError(message, LoadErrorReason.LOADING_DIRECTORY) from e

                    # Otherwise, we don't know the reason, so just raise
                    raise

                if file_path in included:
                    include_provenance = includes_node.get_provenance()
                    raise LoadError(
                        "{}: trying to recursively include {}".format(include_provenance, file_path),
                        LoadErrorReason.RECURSIVE_INCLUDE,
                    )
                # Because the included node will be modified, we need
                # to copy it so that we do not modify the toplevel
                # node of the provenance.
                include_node = include_node.clone()

                try:
                    included.add(file_path)
                    self._process(
                        include_node,
                        included=included,
                        current_loader=sub_loader,
                        only_local=only_local,
                        first_pass=first_pass,
                        process_options=process_options,
                    )
                finally:
                    included.remove(file_path)

                include_node._composite_under(node)

        for value in node.values():
            self._process_value(
                value,
                included=included,
                current_loader=current_loader,
                only_local=only_local,
                first_pass=first_pass,
                process_options=process_options,
            )

    # _include_file()
    #
    # Load include YAML file from with a loader.
    #
    # Args:
    #    include (str): file path relative to loader's project directory.
    #                   Can be prefixed with junctio name.
    #    loader (Loader): Loader for the current project.
    def _include_file(self, include, loader):
        shortname = include
        if ":" in include:
            junction, include = include.split(":", 1)
            current_loader = loader.get_loader(junction)
            current_loader.project.ensure_fully_loaded()
        else:
            current_loader = loader
        project = current_loader.project
        directory = project.directory
        file_path = os.path.join(directory, include)
        key = (current_loader, file_path)
        if key not in self._loaded:
            self._loaded[key] = _yaml.load(file_path, shortname=shortname, project=project, copy_tree=self._copy_tree)
        return self._loaded[key], file_path, current_loader

    # _process_value()
    #
    # Select processing for value that could be a list or a dictionary.
    #
    # Args:
    #    value: Value to process. Can be a list or a dictionary.
    #    included (set): Fail for recursion if trying to load any files in this set
    #    current_loader (Loader): Use alternative loader (for junction files)
    #    only_local (bool): Whether to ignore junction files
    #    first_pass (bool): Whether to use first pass options.
    #    process_options (bool): Whether to process options.
    def _process_value(
        self, value, *, included=None, current_loader=None, only_local=False, first_pass=False, process_options=True,
    ):
        value_type = type(value)

        if value_type is MappingNode:
            self._process_node(
                value,
                included=included,
                current_loader=current_loader,
                only_local=only_local,
                first_pass=first_pass,
                process_options=process_options,
            )
        elif value_type is SequenceNode:
            for v in value:
                self._process_value(
                    v,
                    included=included,
                    current_loader=current_loader,
                    only_local=only_local,
                    first_pass=first_pass,
                    process_options=process_options,
                )