meerschaum.utils.yaml

Meerschaum wrapper around YAML libraries.

This is so switching between PyYAML and ruamel.yaml is smoother.

  1#! /usr/bin/env python
  2# -*- coding: utf-8 -*-
  3# vim:fenc=utf-8
  4
  5"""
  6Meerschaum wrapper around YAML libraries.
  7
  8This is so switching between PyYAML and ruamel.yaml is smoother.
  9"""
 10
 11from meerschaum.utils.misc import filter_keywords
 12from meerschaum.utils.packages import attempt_import, all_packages, _import_module
 13from meerschaum.utils.warnings import error, warn
 14from meerschaum.utils.threading import Lock
 15
 16_lib = None
 17### Also supports 'ruamel.yaml'.
 18_import_name = 'yaml'
 19
 20_yaml = None
 21_locks = {
 22    '_lib': Lock(),
 23    '_yaml': Lock(),
 24}
 25
 26__pdoc__ = {
 27    'attempt_import': False,
 28    'error': False,
 29    'all_packages': False,
 30}
 31
 32
 33def _string_presenter(dumper, data: str):
 34    """
 35    Format strings with newlines as blocks.
 36    https://stackoverflow.com/a/33300001/9699829
 37    """
 38    tag_str = 'tag:yaml.org,2002:str'
 39    kw = {}
 40    if len(data.splitlines()) > 1:
 41        kw['style'] = '|'
 42    return dumper.represent_scalar(tag_str, data, **kw)
 43
 44
 45class yaml:
 46    """
 47    Wrapper around `PyYAML` and `ruamel.yaml` so that we may switch between implementations.
 48    """
 49    global _yaml, _lib
 50    if _import_name is None:
 51        error(f"No YAML library declared.")
 52    with _locks['_lib']:
 53        try:
 54            _lib = _import_module(_import_name)
 55        except (ImportError, ModuleNotFoundError):
 56            _lib = attempt_import(_import_name, split=False, lazy=False, install=True)
 57    with _locks['_yaml']:
 58        _yaml = _lib if _import_name != 'ruamel.yaml' else _lib.YAML()
 59        if _import_name != 'ruamel.yaml':
 60            _yaml.add_representer(str, _string_presenter)
 61            _yaml.representer.SafeRepresenter.add_representer(str, _string_presenter)
 62
 63
 64    @staticmethod
 65    def safe_load(*args, **kw):
 66        """
 67        Execute `safe_load` for `PyYAML` and `load` for `ruamel.yaml`.
 68        """
 69        if _import_name == 'ruamel.yaml':
 70            return _yaml.load(*args, **filter_keywords(_yaml.load, **kw))
 71        return _yaml.safe_load(*args, **filter_keywords(_yaml.safe_load, **kw))
 72
 73
 74    @staticmethod
 75    def load(*args, **kw):
 76        """
 77        Execute `yaml.load()`.
 78        Handles the breaking change at `v6.0` of `PyYAML`
 79        (added `yaml.Loader` as a positional argument).
 80        """
 81        packaging_version = attempt_import('packaging.version')
 82        _args = list(args)
 83        if (
 84            _import_name == 'yaml'
 85            and packaging_version.parse(_yaml.__version__) >= packaging_version.parse('6.0')
 86        ):
 87            _args += [_yaml.Loader]
 88        return _yaml.load(*_args, **filter_keywords(_yaml.load, **kw))
 89
 90
 91    @staticmethod
 92    def dump(data, stream=None, **kw):
 93        """
 94        Dump to a stream. If no stream is provided, return a string instead.
 95        For `ruamel.yaml`, it dumps into a `StringIO` stream and returns `getvalue()`.
 96        """
 97        get_string = False
 98        if stream is None and _import_name == 'ruamel.yaml':
 99            stream = _lib.compat.StringIO()
100            get_string = True
101        result = _yaml.dump(data, stream, **filter_keywords(_yaml.dump, **kw))
102        if get_string:
103            return stream.getvalue()
104        return result
class yaml:
 46class yaml:
 47    """
 48    Wrapper around `PyYAML` and `ruamel.yaml` so that we may switch between implementations.
 49    """
 50    global _yaml, _lib
 51    if _import_name is None:
 52        error(f"No YAML library declared.")
 53    with _locks['_lib']:
 54        try:
 55            _lib = _import_module(_import_name)
 56        except (ImportError, ModuleNotFoundError):
 57            _lib = attempt_import(_import_name, split=False, lazy=False, install=True)
 58    with _locks['_yaml']:
 59        _yaml = _lib if _import_name != 'ruamel.yaml' else _lib.YAML()
 60        if _import_name != 'ruamel.yaml':
 61            _yaml.add_representer(str, _string_presenter)
 62            _yaml.representer.SafeRepresenter.add_representer(str, _string_presenter)
 63
 64
 65    @staticmethod
 66    def safe_load(*args, **kw):
 67        """
 68        Execute `safe_load` for `PyYAML` and `load` for `ruamel.yaml`.
 69        """
 70        if _import_name == 'ruamel.yaml':
 71            return _yaml.load(*args, **filter_keywords(_yaml.load, **kw))
 72        return _yaml.safe_load(*args, **filter_keywords(_yaml.safe_load, **kw))
 73
 74
 75    @staticmethod
 76    def load(*args, **kw):
 77        """
 78        Execute `yaml.load()`.
 79        Handles the breaking change at `v6.0` of `PyYAML`
 80        (added `yaml.Loader` as a positional argument).
 81        """
 82        packaging_version = attempt_import('packaging.version')
 83        _args = list(args)
 84        if (
 85            _import_name == 'yaml'
 86            and packaging_version.parse(_yaml.__version__) >= packaging_version.parse('6.0')
 87        ):
 88            _args += [_yaml.Loader]
 89        return _yaml.load(*_args, **filter_keywords(_yaml.load, **kw))
 90
 91
 92    @staticmethod
 93    def dump(data, stream=None, **kw):
 94        """
 95        Dump to a stream. If no stream is provided, return a string instead.
 96        For `ruamel.yaml`, it dumps into a `StringIO` stream and returns `getvalue()`.
 97        """
 98        get_string = False
 99        if stream is None and _import_name == 'ruamel.yaml':
100            stream = _lib.compat.StringIO()
101            get_string = True
102        result = _yaml.dump(data, stream, **filter_keywords(_yaml.dump, **kw))
103        if get_string:
104            return stream.getvalue()
105        return result

Wrapper around PyYAML and ruamel.yaml so that we may switch between implementations.

@staticmethod
def safe_load(*args, **kw):
65    @staticmethod
66    def safe_load(*args, **kw):
67        """
68        Execute `safe_load` for `PyYAML` and `load` for `ruamel.yaml`.
69        """
70        if _import_name == 'ruamel.yaml':
71            return _yaml.load(*args, **filter_keywords(_yaml.load, **kw))
72        return _yaml.safe_load(*args, **filter_keywords(_yaml.safe_load, **kw))

Execute safe_load for PyYAML and load for ruamel.yaml.

@staticmethod
def load(*args, **kw):
75    @staticmethod
76    def load(*args, **kw):
77        """
78        Execute `yaml.load()`.
79        Handles the breaking change at `v6.0` of `PyYAML`
80        (added `yaml.Loader` as a positional argument).
81        """
82        packaging_version = attempt_import('packaging.version')
83        _args = list(args)
84        if (
85            _import_name == 'yaml'
86            and packaging_version.parse(_yaml.__version__) >= packaging_version.parse('6.0')
87        ):
88            _args += [_yaml.Loader]
89        return _yaml.load(*_args, **filter_keywords(_yaml.load, **kw))

Execute yaml.load(). Handles the breaking change at v6.0 of PyYAML (added yaml.Loader as a positional argument).

@staticmethod
def dump(data, stream=None, **kw):
 92    @staticmethod
 93    def dump(data, stream=None, **kw):
 94        """
 95        Dump to a stream. If no stream is provided, return a string instead.
 96        For `ruamel.yaml`, it dumps into a `StringIO` stream and returns `getvalue()`.
 97        """
 98        get_string = False
 99        if stream is None and _import_name == 'ruamel.yaml':
100            stream = _lib.compat.StringIO()
101            get_string = True
102        result = _yaml.dump(data, stream, **filter_keywords(_yaml.dump, **kw))
103        if get_string:
104            return stream.getvalue()
105        return result

Dump to a stream. If no stream is provided, return a string instead. For ruamel.yaml, it dumps into a StringIO stream and returns getvalue().