Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.util.yaml_util

import logging
import os
from collections import OrderedDict

import yaml
from yaml.constructor import ConstructorError

try:
    from yaml import CSafeLoader as SafeLoader
except ImportError:
    from yaml import SafeLoader  # type: ignore[misc]


log = logging.getLogger(__name__)


[docs]class OrderedLoader(SafeLoader): # This class was pulled out of ordered_load() for the sake of # mocking __init__ in a unit test.
[docs] def __init__(self, stream): self._root = os.path.split(stream.name)[0] super().__init__(stream)
[docs] def include(self, node): filename = os.path.join(self._root, self.construct_scalar(node)) with open(filename) as f: return yaml.load(f, OrderedLoader)
[docs]def ordered_load(stream, merge_duplicate_keys=False): """ Parse the first YAML document in a stream and produce the corresponding Python object. If merge_duplicate_keys is True, merge the values of duplicate mapping keys into a list, as the uWSGI "dumb" YAML parser would do. Otherwise, following YAML 1.2 specification which says that "each key is unique in the association", raise a ConstructionError exception. """ def construct_mapping(loader, node, deep=False): loader.flatten_mapping(node) mapping = {} merged_duplicate = {} for key_node, value_node in node.value: key = loader.construct_object(key_node, deep=deep) value = loader.construct_object(value_node, deep=deep) if key in mapping: if not merge_duplicate_keys: raise ConstructorError( "while constructing a mapping", node.start_mark, f"found duplicated key ({key})", key_node.start_mark, ) log.debug("Merging values for duplicate key '%s' into a list", key) if merged_duplicate.get(key): mapping[key].append(value) else: mapping[key] = [mapping[key], value] merged_duplicate[key] = True else: mapping[key] = value return mapping OrderedLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping) OrderedLoader.add_constructor("!include", OrderedLoader.include) return yaml.load(stream, OrderedLoader)
[docs]def ordered_dump(data, stream=None, Dumper=yaml.Dumper, **kwds): class OrderedDumper(Dumper): pass def _dict_representer(dumper, data): return dumper.represent_mapping(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, list(data.items())) OrderedDumper.add_representer(OrderedDict, _dict_representer) return yaml.dump(data, stream, OrderedDumper, **kwds)