ustr_representer() - Code Metrics - Inspection of "fix: [yaml] add one more workaround for python 3" - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e8b00e...f4beed )

by Satoru

created 2017-03-12 05:12 UTC

ustr_representer() A

↳ Parent: _customized_dumper()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	2	Features	2

Metric	Value
cc	1
c	2
b	2
f	2
dl	0
loc	3
rs	10

#
# Copyright (C) 2011 - 2017 Satoru SATOH <ssato @ redhat.com>
# License: MIT
#
# type() is used to exactly match check instead of isinstance here.
# pylint: disable=unidiomatic-typecheck
r"""YAML backend:

- Format to support: YAML, http://yaml.org
- Requirements: PyYAML (yaml), http://pyyaml.org
- Development Status :: 5 - Production/Stable
- Limitations:

  - Resuls is not ordered even if 'ac_ordered' or 'ac_dict' was given.

- Special options:

  - All keyword options of yaml.safe_load, yaml.load, yaml.safe_dump and
    yaml.dump should work.

  - Use 'ac_safe' boolean keyword option if you prefer to call yaml.safe_load
    and yaml.safe_dump instead of yaml.load and yaml.dump. Please note that
    this option conflicts with 'ac_dict' option and these options cannot be
    used at the same time.

  - See also: http://pyyaml.org/wiki/PyYAMLDocumentation

Changelog:

.. versionchanged:: 0.3

   - Changed special keyword option 'ac_safe' from 'safe' to avoid
     possibility of option conflicts in the future.
"""
from __future__ import absolute_import

import yaml
try:
    from yaml import CSafeLoader as Loader, CDumper as Dumper
except ImportError:
    from yaml import SafeLoader as Loader, Dumper

import anyconfig.backend.base
import anyconfig.compat
import anyconfig.utils


_MAPPING_TAG = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG


def _filter_from_options(key, options):
    """
    :param key: Key str in options
    :param options: Mapping object
    :return:
        New mapping object from `options` in which the item with `key` filtered

    >>> _filter_from_options('a', dict(a=1, b=2))
    {'b': 2}
    """
    return anyconfig.utils.filter_options([k for k in options.keys()
                                           if k != key], options)


def _customized_loader(container, loader=Loader, mapping_tag=_MAPPING_TAG):
    """
    Create or update loader with making given callble `container` to make
    mapping objects such as dict and OrderedDict, used to construct python
    object from yaml mapping node internally.

    :param container: Set container used internally
    """
    def construct_mapping(loader, node, deep=False):
        """Construct python object from yaml mapping node, based on
        :meth:`yaml.BaseConstructor.construct_mapping` in PyYAML (MIT).
        """
        if not isinstance(node, yaml.MappingNode):
            msg = "expected a mapping node, but found %s" % node.id
            raise yaml.constructor.ConstructorError(None, None, msg,
                                                    node.start_mark)
        mapping = container()
        for key_node, value_node in node.value:
            key = loader.construct_object(key_node, deep=deep)
            try:
                hash(key)
            except TypeError as exc:
                eargs = ("while constructing a mapping",
                         node.start_mark,
                         "found unacceptable key (%s)" % exc,
                         key_node.start_mark)
                raise yaml.constructor.ConstructorError(*eargs)
            value = loader.construct_object(value_node, deep=deep)
            mapping[key] = value

        return mapping

    if type(container) != dict:
        loader.add_constructor(mapping_tag, construct_mapping)
    return loader


def _customized_dumper(container, dumper=Dumper):
    """
    Coutnerpart of :func:`_customized_loader` for dumpers.
    """
    def container_representer(dumper, data, mapping_tag=_MAPPING_TAG):
        """Container representer.
        """
        return dumper.represent_mapping(mapping_tag, data.items())

    def ustr_representer(dumper, data):
        tag = "tag:yaml.org,2002:python/unicode"
        return dumper.represent_scalar(tag, data)

    try:
        dumper.add_representer(unicode, ustr_representer)
    except NameError:
        pass

    if type(container) != dict:
        dumper.add_representer(container, container_representer)
    return dumper


def _yml_fnc(fname, *args, **options):
    """An wrapper of yaml.safe_load, yaml.load, yaml.safe_dump and yaml.dump.

    :param fname:
        "load" or "dump", not checked but it should be OK.
        see also :func:`_yml_load` and :func:`_yml_dump`
    :param args: [stream] for load or [cnf, stream] for dump
    :param options: keyword args may contain "ac_safe" to load/dump safely
    """
    key = "ac_safe"
    fnc = getattr(yaml, r"safe_" + fname if options.get(key) else fname)
    return fnc(*args, **_filter_from_options(key, options))


def _yml_load(stream, container, **options):
    """An wrapper of yaml.safe_load and yaml.load.

    :param stream: a file or file-like object to load YAML content
    :param container: callble to make a container object

    :return: Mapping object
    """
    if options.get("ac_safe", False):
        options = {}  # yaml.safe_load does not process Loader opts.
    elif not options.get("Loader"):
        maybe_container = options.get("ac_dict", False)
        if maybe_container and callable(maybe_container):
            container = maybe_container

        options["Loader"] = _customized_loader(container)

    ret = _yml_fnc("load", stream, **_filter_from_options("ac_dict", options))
    return container() if ret is None else container(ret)


def _yml_dump(cnf, stream, **options):
    """An wrapper of yaml.safe_dump and yaml.dump.

    :param cnf: Mapping object to dump
    :param stream: a file or file-like object to dump YAML data
    """
    if options.get("ac_safe", False):
        options = {}
    elif not options.get("Dumper", False):
        # TODO: Any other way to get its constructor?
        cnf_type = type(cnf)
        maybe_container = options.get("ac_dict", cnf_type)
        options["Dumper"] = _customized_dumper(maybe_container)

    # Type information and the order of items are lost on dump currently.
    cnf = anyconfig.dicts.convert_to(cnf, ac_dict=dict)
    options = _filter_from_options("ac_dict", options)
    return _yml_fnc("dump", cnf, stream, **options)


class Parser(anyconfig.backend.base.FromStreamLoader,
             anyconfig.backend.base.ToStreamDumper):
    """
    Parser for YAML files.
    """
    _type = "yaml"
    _extensions = ["yaml", "yml"]
    _load_opts = ["Loader", "ac_safe", "ac_dict"]
    _dump_opts = ["stream", "ac_safe", "Dumper", "default_style",
                  "default_flow_style", "canonical", "indent", "width",
                  "allow_unicode", "line_break", "encoding", "explicit_start",
                  "explicit_end", "version", "tags"]
    _ordered = True
    _dict_opts = ["ac_dict"]

    load_from_stream = anyconfig.backend.base.to_method(_yml_load)
    dump_to_stream = anyconfig.backend.base.to_method(_yml_dump)

# vim:sw=4:ts=4:et:


1			#
2			# Copyright (C) 2011 - 2017 Satoru SATOH <ssato @ redhat.com>
3			# License: MIT
4			#
5			# type() is used to exactly match check instead of isinstance here.
6			# pylint: disable=unidiomatic-typecheck
7			r"""YAML backend:
8
9			- Format to support: YAML, http://yaml.org
10			- Requirements: PyYAML (yaml), http://pyyaml.org
11			- Development Status :: 5 - Production/Stable
12			- Limitations:
13
14			- Resuls is not ordered even if 'ac_ordered' or 'ac_dict' was given.
15
16			- Special options:
17
18			- All keyword options of yaml.safe_load, yaml.load, yaml.safe_dump and
19			yaml.dump should work.
20
21			- Use 'ac_safe' boolean keyword option if you prefer to call yaml.safe_load
22			and yaml.safe_dump instead of yaml.load and yaml.dump. Please note that
23			this option conflicts with 'ac_dict' option and these options cannot be
24			used at the same time.
25
26			- See also: http://pyyaml.org/wiki/PyYAMLDocumentation
27
28			Changelog:
29
30			.. versionchanged:: 0.3
31
32			- Changed special keyword option 'ac_safe' from 'safe' to avoid
33			possibility of option conflicts in the future.
34			"""
35			from __future__ import absolute_import
36
37			import yaml
38			try:
39			from yaml import CSafeLoader as Loader, CDumper as Dumper
40			except ImportError:
41			from yaml import SafeLoader as Loader, Dumper
42
43			import anyconfig.backend.base
44			import anyconfig.compat
45			import anyconfig.utils
46
47
48			_MAPPING_TAG = yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
49
50
51			def _filter_from_options(key, options):
52			"""
53			:param key: Key str in options
54			:param options: Mapping object
55			:return:
56			New mapping object from `options` in which the item with `key` filtered
57
58			>>> _filter_from_options('a', dict(a=1, b=2))
59			{'b': 2}
60			"""
61			return anyconfig.utils.filter_options([k for k in options.keys()
62			if k != key], options)
63
64
65			def _customized_loader(container, loader=Loader, mapping_tag=_MAPPING_TAG):
66			"""
67			Create or update loader with making given callble `container` to make
68			mapping objects such as dict and OrderedDict, used to construct python
69			object from yaml mapping node internally.
70
71			:param container: Set container used internally
72			"""
73			def construct_mapping(loader, node, deep=False):
74			"""Construct python object from yaml mapping node, based on
75			:meth:`yaml.BaseConstructor.construct_mapping` in PyYAML (MIT).
76			"""
77			if not isinstance(node, yaml.MappingNode):
78			msg = "expected a mapping node, but found %s" % node.id
79			raise yaml.constructor.ConstructorError(None, None, msg,
80			node.start_mark)
81			mapping = container()
82			for key_node, value_node in node.value:
83			key = loader.construct_object(key_node, deep=deep)
84			try:
85			hash(key)
86			except TypeError as exc:
87			eargs = ("while constructing a mapping",
88			node.start_mark,
89			"found unacceptable key (%s)" % exc,
90			key_node.start_mark)
91			raise yaml.constructor.ConstructorError(*eargs)
92			value = loader.construct_object(value_node, deep=deep)
93			mapping[key] = value
94
95			return mapping
96
97			if type(container) != dict:
98			loader.add_constructor(mapping_tag, construct_mapping)
99			return loader
100
101
102			def _customized_dumper(container, dumper=Dumper):
103			"""
104			Coutnerpart of :func:`_customized_loader` for dumpers.
105			"""
106			def container_representer(dumper, data, mapping_tag=_MAPPING_TAG):
107			"""Container representer.
108			"""
109			return dumper.represent_mapping(mapping_tag, data.items())
110
111			def ustr_representer(dumper, data):
112			tag = "tag:yaml.org,2002:python/unicode"
113			return dumper.represent_scalar(tag, data)
114
115			try:
116			dumper.add_representer(unicode, ustr_representer)
117			except NameError:
118			pass
119
120			if type(container) != dict:
121			dumper.add_representer(container, container_representer)
122			return dumper
123
124
125			def _yml_fnc(fname, args, *options):
126			"""An wrapper of yaml.safe_load, yaml.load, yaml.safe_dump and yaml.dump.
127
128			:param fname:
129			"load" or "dump", not checked but it should be OK.
130			see also :func:`_yml_load` and :func:`_yml_dump`
131			:param args: [stream] for load or [cnf, stream] for dump
132			:param options: keyword args may contain "ac_safe" to load/dump safely
133			"""
134			key = "ac_safe"
135			fnc = getattr(yaml, r"safe_" + fname if options.get(key) else fname)
136			return fnc(args, *_filter_from_options(key, options))
137
138
139			def _yml_load(stream, container, **options):
140			"""An wrapper of yaml.safe_load and yaml.load.
141
142			:param stream: a file or file-like object to load YAML content
143			:param container: callble to make a container object
144
145			:return: Mapping object
146			"""
147			if options.get("ac_safe", False):
148			options = {} # yaml.safe_load does not process Loader opts.
149			elif not options.get("Loader"):
150			maybe_container = options.get("ac_dict", False)
151			if maybe_container and callable(maybe_container):
152			container = maybe_container
153
154			options["Loader"] = _customized_loader(container)
155
156			ret = _yml_fnc("load", stream, **_filter_from_options("ac_dict", options))
157			return container() if ret is None else container(ret)
158
159
160			def _yml_dump(cnf, stream, **options):
161			"""An wrapper of yaml.safe_dump and yaml.dump.
162
163			:param cnf: Mapping object to dump
164			:param stream: a file or file-like object to dump YAML data
165			"""
166			if options.get("ac_safe", False):
167			options = {}
168			elif not options.get("Dumper", False):
169			# TODO: Any other way to get its constructor?
170			cnf_type = type(cnf)
171			maybe_container = options.get("ac_dict", cnf_type)
172			options["Dumper"] = _customized_dumper(maybe_container)
173
174			# Type information and the order of items are lost on dump currently.
175			cnf = anyconfig.dicts.convert_to(cnf, ac_dict=dict)
176			options = _filter_from_options("ac_dict", options)
177			return _yml_fnc("dump", cnf, stream, **options)
178
179
180			class Parser(anyconfig.backend.base.FromStreamLoader,
181			anyconfig.backend.base.ToStreamDumper):
182			"""
183			Parser for YAML files.
184			"""
185			_type = "yaml"
186			_extensions = ["yaml", "yml"]
187			_load_opts = ["Loader", "ac_safe", "ac_dict"]
188			_dump_opts = ["stream", "ac_safe", "Dumper", "default_style",
189			"default_flow_style", "canonical", "indent", "width",
190			"allow_unicode", "line_break", "encoding", "explicit_start",
191			"explicit_end", "version", "tags"]
192			_ordered = True
193			_dict_opts = ["ac_dict"]
194
195			load_from_stream = anyconfig.backend.base.to_method(_yml_load)
196			dump_to_stream = anyconfig.backend.base.to_method(_yml_dump)
197
198			# vim:sw=4:ts=4:et:
199

ssato / python-anyconfig

Push — master ( e8b00e...f4beed )

ustr_representer() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like