Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/galaxy/util/yaml_util.py @ 0:26e78fe6e8c4 draft
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
author | shellac |
---|---|
date | Sat, 02 May 2020 07:14:21 -0400 |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.7/site-packages/galaxy/util/yaml_util.py Sat May 02 07:14:21 2020 -0400 @@ -0,0 +1,82 @@ +from __future__ import absolute_import + +import logging +import os +from collections import OrderedDict + +import yaml +from yaml.constructor import ConstructorError + + +log = logging.getLogger(__name__) + +OPTION_DEFAULTS = { + "type": "str", + "unknown_option": False, + "default": None, + "desc": None, +} + + +class OrderedLoader(yaml.SafeLoader): + # This class was pulled out of ordered_load() for the sake of + # mocking __init__ in a unit test. + def __init__(self, stream): + self._root = os.path.split(stream.name)[0] + super(OrderedLoader, self).__init__(stream) + + def include(self, node): + filename = os.path.join(self._root, self.construct_scalar(node)) + with open(filename, 'r') as f: + return yaml.load(f, OrderedLoader) + + +def ordered_load(stream, merge_duplicate_keys=False): + """ + Parse the first YAML document in a stream and produce the corresponding + Python object, using OrderedDicts instead of dicts. + + 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 = OrderedDict() + 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, + "found duplicated key (%s)" % 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) + + +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)