construct_mapping() - Code Metrics - Inspection of "fix: [yaml] add workaround to add constructor for..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( f4beed...16e52d )

by Satoru

created 2017-03-12 06:37 UTC

construct_mapping() B

↳ Parent: _customized_loader()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	1	Features	0

Metric	Value
cc	4
c	3
b	1
f	0
dl	0
loc	23
rs	8.7972

#
# 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

    tag = "tag:yaml.org,2002:python/unicode"

    def construct_ustr(loader, node):
        return loader.construct_scalar(node)

    try:
        loader.add_constructor(tag, construct_ustr)
    except NameError:
        pass

    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			tag = "tag:yaml.org,2002:python/unicode"
98
99			def construct_ustr(loader, node):
100			return loader.construct_scalar(node)
101
102			try:
103			loader.add_constructor(tag, construct_ustr)
104			except NameError:
105			pass
106
107			if type(container) != dict:
108			loader.add_constructor(mapping_tag, construct_mapping)
109			return loader
110
111
112			def _customized_dumper(container, dumper=Dumper):
113			"""
114			Coutnerpart of :func:`_customized_loader` for dumpers.
115			"""
116			def container_representer(dumper, data, mapping_tag=_MAPPING_TAG):
117			"""Container representer.
118			"""
119			return dumper.represent_mapping(mapping_tag, data.items())
120
121			def ustr_representer(dumper, data):
122			tag = "tag:yaml.org,2002:python/unicode"
123			return dumper.represent_scalar(tag, data)
124
125			try:
126			dumper.add_representer(unicode, ustr_representer)
127			except NameError:
128			pass
129
130			if type(container) != dict:
131			dumper.add_representer(container, container_representer)
132			return dumper
133
134
135			def _yml_fnc(fname, args, *options):
136			"""An wrapper of yaml.safe_load, yaml.load, yaml.safe_dump and yaml.dump.
137
138			:param fname:
139			"load" or "dump", not checked but it should be OK.
140			see also :func:`_yml_load` and :func:`_yml_dump`
141			:param args: [stream] for load or [cnf, stream] for dump
142			:param options: keyword args may contain "ac_safe" to load/dump safely
143			"""
144			key = "ac_safe"
145			fnc = getattr(yaml, r"safe_" + fname if options.get(key) else fname)
146			return fnc(args, *_filter_from_options(key, options))
147
148
149			def _yml_load(stream, container, **options):
150			"""An wrapper of yaml.safe_load and yaml.load.
151
152			:param stream: a file or file-like object to load YAML content
153			:param container: callble to make a container object
154
155			:return: Mapping object
156			"""
157			if options.get("ac_safe", False):
158			options = {} # yaml.safe_load does not process Loader opts.
159			elif not options.get("Loader"):
160			maybe_container = options.get("ac_dict", False)
161			if maybe_container and callable(maybe_container):
162			container = maybe_container
163
164			options["Loader"] = _customized_loader(container)
165
166			ret = _yml_fnc("load", stream, **_filter_from_options("ac_dict", options))
167			return container() if ret is None else container(ret)
168
169
170			def _yml_dump(cnf, stream, **options):
171			"""An wrapper of yaml.safe_dump and yaml.dump.
172
173			:param cnf: Mapping object to dump
174			:param stream: a file or file-like object to dump YAML data
175			"""
176			if options.get("ac_safe", False):
177			options = {}
178			elif not options.get("Dumper", False):
179			# TODO: Any other way to get its constructor?
180			cnf_type = type(cnf)
181			maybe_container = options.get("ac_dict", cnf_type)
182			options["Dumper"] = _customized_dumper(maybe_container)
183
184			# Type information and the order of items are lost on dump currently.
185			cnf = anyconfig.dicts.convert_to(cnf, ac_dict=dict)
186			options = _filter_from_options("ac_dict", options)
187			return _yml_fnc("dump", cnf, stream, **options)
188
189
190			class Parser(anyconfig.backend.base.FromStreamLoader,
191			anyconfig.backend.base.ToStreamDumper):
192			"""
193			Parser for YAML files.
194			"""
195			_type = "yaml"
196			_extensions = ["yaml", "yml"]
197			_load_opts = ["Loader", "ac_safe", "ac_dict"]
198			_dump_opts = ["stream", "ac_safe", "Dumper", "default_style",
199			"default_flow_style", "canonical", "indent", "width",
200			"allow_unicode", "line_break", "encoding", "explicit_start",
201			"explicit_end", "version", "tags"]
202			_ordered = True
203			_dict_opts = ["ac_dict"]
204
205			load_from_stream = anyconfig.backend.base.to_method(_yml_load)
206			dump_to_stream = anyconfig.backend.base.to_method(_yml_dump)
207
208			# vim:sw=4:ts=4:et:
209

ssato / python-anyconfig

Push — master ( f4beed...16e52d )

construct_mapping() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like