| import os |
| from collections.abc import Mapping |
| from . import _yaml |
| from ._exceptions import LoadError, 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 |
| # process_project_options (bool): Whether to process options from current project |
| # |
| def process(self, node, *, only_local=False, process_project_options=True): |
| self._process(node, only_local=only_local, process_project_options=process_project_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 |
| # process_project_options (bool): Whether to process options from current project |
| # |
| def _process(self, node, *, included=set(), current_loader=None, only_local=False, process_project_options=True): |
| |
| if current_loader is None: |
| current_loader = self._loader |
| |
| if process_project_options: |
| current_loader.project.options.process_node(node) |
| |
| self._process_node( |
| node, |
| included=included, |
| only_local=only_local, |
| current_loader=current_loader, |
| process_project_options=process_project_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 |
| # process_project_options (bool): Whether to process options from current project |
| # |
| def _process_node( |
| self, node, *, included=set(), current_loader=None, only_local=False, process_project_options=True |
| ): |
| |
| if isinstance(node.get('(@)'), str): |
| includes = [_yaml.node_get(node, str, '(@)')] |
| else: |
| includes = _yaml.node_get(node, list, '(@)', default_value=None) |
| if '(@)' in node: |
| del node['(@)'] |
| |
| if includes: |
| for include in reversed(includes): |
| if only_local and ':' in include: |
| continue |
| include_node, file_path, sub_loader = self._include_file(include, |
| current_loader) |
| if file_path in included: |
| provenance = _yaml.node_get_provenance(node) |
| raise LoadError(LoadErrorReason.RECURSIVE_INCLUDE, |
| "{}: trying to recursively include {}". format(provenance, |
| file_path)) |
| # 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 = _yaml.node_chain_copy(include_node) |
| |
| try: |
| included.add(file_path) |
| self._process( |
| include_node, included=included, |
| current_loader=sub_loader, |
| only_local=only_local, |
| process_project_options=process_project_options or current_loader != sub_loader, |
| ) |
| finally: |
| included.remove(file_path) |
| |
| _yaml.composite(include_node, node) |
| to_delete = [key for key, _ in _yaml.node_items(node) if key not in include_node] |
| for key, value in include_node.items(): |
| node[key] = value |
| for key in to_delete: |
| del node[key] |
| |
| for _, value in _yaml.node_items(node): |
| self._process_value( |
| value, |
| included=included, |
| current_loader=current_loader, |
| only_local=only_local, |
| process_project_options=process_project_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) |
| junction_loader = loader._get_loader(junction, fetch_subprojects=True) |
| current_loader = junction_loader |
| 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(os.path.join(directory, include), |
| 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 |
| # process_project_options (bool): Whether to process options from current project |
| # |
| def _process_value( |
| self, value, *, included=set(), current_loader=None, only_local=False, process_project_options=True |
| ): |
| if isinstance(value, Mapping): |
| self._process_node( |
| value, |
| included=included, |
| current_loader=current_loader, |
| only_local=only_local, |
| process_project_options=process_project_options, |
| ) |
| elif isinstance(value, list): |
| for v in value: |
| self._process_value( |
| v, |
| included=included, |
| current_loader=current_loader, |
| only_local=only_local, |
| process_project_options=process_project_options, |
| ) |