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 # 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 def process(self, node, *, included=set(), current_loader=None, only_local=False): if current_loader is None: current_loader = self._loader 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) 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) # _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 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 def _process_value(self, value, *, included=set(), current_loader=None, only_local=False): if isinstance(value, Mapping): self.process(value, included=included, current_loader=current_loader, only_local=only_local) elif isinstance(value, list): for v in value: self._process_value(v, included=included, current_loader=current_loader, only_local=only_local)