Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/gxformat2/yaml.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
| author | shellac | 
|---|---|
| date | Mon, 22 Mar 2021 18:12:50 +0000 | 
| parents | |
| children | 
   comparison
  equal
  deleted
  inserted
  replaced
| -1:000000000000 | 0:4f3585e2f14b | 
|---|---|
| 1 """YAML loading utilities for gxformat2.""" | |
| 2 from collections import OrderedDict | |
| 3 | |
| 4 try: | |
| 5 from galaxy.model.custom_types import MutationDict # type: ignore | |
| 6 except ImportError: | |
| 7 MutationDict = None | |
| 8 import yaml | |
| 9 | |
| 10 | |
| 11 def ordered_load_path(path: str, **kwds): | |
| 12 """Safe and ordered load of YAML from specified path.""" | |
| 13 with open(path, "r") as f: | |
| 14 return ordered_load(f, **kwds) | |
| 15 | |
| 16 | |
| 17 def ordered_load(stream, Loader=yaml.SafeLoader, **kwds): | |
| 18 """Safe and ordered load of YAML from stream.""" | |
| 19 class OrderedLoader(Loader): | |
| 20 pass | |
| 21 | |
| 22 def construct_mapping(loader, node): | |
| 23 loader.flatten_mapping(node) | |
| 24 return OrderedDict(loader.construct_pairs(node)) | |
| 25 | |
| 26 OrderedLoader.add_constructor( | |
| 27 yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, | |
| 28 construct_mapping) | |
| 29 | |
| 30 return yaml.load(stream, OrderedLoader, **kwds) | |
| 31 | |
| 32 | |
| 33 def ordered_dump(data, stream=None, Dumper=yaml.SafeDumper, **kwds): | |
| 34 """Safe and ordered dump of YAML to stream.""" | |
| 35 class OrderedDumper(Dumper): | |
| 36 pass | |
| 37 | |
| 38 def _dict_representer(dumper, data): | |
| 39 return dumper.represent_mapping( | |
| 40 yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, | |
| 41 list(data.items())) | |
| 42 OrderedDumper.add_representer(OrderedDict, _dict_representer) | |
| 43 if MutationDict is not None: | |
| 44 OrderedDumper.add_representer(MutationDict, _dict_representer) | |
| 45 | |
| 46 return yaml.dump(data, stream, OrderedDumper, **kwds) | |
| 47 | |
| 48 | |
| 49 def ordered_dump_to_path(as_dict: dict, path: str): | |
| 50 """Safe and ordered dump of YAML to path.""" | |
| 51 with open(path, "w") as f: | |
| 52 ordered_dump(as_dict, f) | |
| 53 | |
| 54 | |
| 55 __all__ = ( | |
| 56 'ordered_dump', | |
| 57 'ordered_dump_to_path', | |
| 58 'ordered_load', | |
| 59 'ordered_load_path', | |
| 60 ) | 
