anyconfig._maybe_validated() - Code Metrics - Inspection of "enhancement: add a test case of dump & multi_load..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( c877c2...3e5906 )

by Satoru

created 2015-12-16 11:16 UTC

anyconfig._maybe_validated() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	5
dl	0
loc	21
rs	8.439

#
# Copyright (C) 2012 - 2015 Satoru SATOH <ssato @ redhat.com>
# License: MIT
#
# pylint: disable=unused-import,import-error,invalid-name
r"""Public APIs of anyconfig module.

.. versionadded:: 0.4.99
   Added ac_namedtuple parameter to \*load and \*dump APIs.

.. versionchanged:: 0.3
   Replaced `forced_type` optional argument of some public APIs with
   `ac_parser` to allow skip of config parser search by passing parser object
   previously found and instantiated.

   Also removed some optional arguments, `ignore_missing`, `merge` and
   `marker`, from definitions of some public APIs as these may not be changed
   from default in common use cases.

.. versionchanged:: 0.2
   Now APIs :func:`find_loader`, :func:`single_load`, :func:`multi_load`,
   :func:`load` and :func:`dump` can process a file/file-like object or a list
   of file/file-like objects instead of a file path or a list of file paths.

.. versionadded:: 0.2
   Export factory method (create) of anyconfig.mergeabledict.MergeableDict
"""
from __future__ import absolute_import

import os.path

from anyconfig.globals import LOGGER
import anyconfig.backends
import anyconfig.backend.json
import anyconfig.compat
import anyconfig.mergeabledict
import anyconfig.parser
import anyconfig.template
import anyconfig.utils

# Import some global constants will be re-exported:
from anyconfig.mergeabledict import (
    MS_REPLACE, MS_NO_REPLACE, MS_DICTS, MS_DICTS_AND_LISTS, MERGE_STRATEGIES,
    PATH_SEPS, get, set_, convert_to, create_from  # flake8: noqa
)
from anyconfig.schema import validate, gen_schema
from anyconfig.utils import is_path

# Re-export and aliases:
list_types = anyconfig.backends.list_types  # flake8: noqa
container = anyconfig.mergeabledict.MergeableDict
to_container = container.create


def _is_paths(maybe_paths):
    """
    Does given object `maybe_paths` consist of path or path pattern strings?
    """
    if anyconfig.utils.is_iterable(maybe_paths):
        return not getattr(maybe_paths, "read", False)

    return False  # Not an iterable at least.


def _maybe_validated(cnf, schema, format_checker=None, ac_namedtuple=False):
    """
    :param cnf: Configuration object :: container
    :param schema: JSON schema object :: container
    :param format_checker: A format property checker object of which class is
        inherited from jsonschema.FormatChecker, it's default if None given.
    :param ac_namedtuple: Convert result to nested namedtuple object if True

    :return: Given `cnf` as it is if validation succeeds else None
    """
    if schema is None:
        valid = True
    else:
        (valid, msg) = validate(cnf, schema, format_checker, True)
        if msg:
            LOGGER.warning(msg)

    if valid:
        return convert_to(cnf, True) if ac_namedtuple else cnf

    return None


def set_loglevel(level):
    """
    :param level: Log level, e.g. logging.INFO and logging.WARN.
    """
    LOGGER.setLevel(level)


def find_loader(path_or_stream, parser_or_type=None, is_path_=None):
    """
    Find out config parser object appropriate to load from a file of given path
    or file/file-like object.

    :param path_or_stream: Configuration file path or file / file-like object
    :param parser_or_type: Forced configuration parser type or parser object
    :param is_path_: True if given `path_or_stream` is a file path

    :return: Config parser instance or None
    """
    if anyconfig.backends.is_parser(parser_or_type):
        return parser_or_type

    (psr, err) = anyconfig.backends.find_parser(path_or_stream, parser_or_type,
                                                is_path_=is_path_)
    if psr is None:
        LOGGER.error(err)
        return None

    LOGGER.debug("Using config parser: %r [%s]", psr, psr.type())
    return psr()  # TBD: Passing initialization arguments.


def single_load(path_or_stream, ac_parser=None, ac_template=False,
                ac_context=None, ac_schema=None, ac_namedtuple=False,
                **kwargs):
    """
    Load single config file.

    :param path_or_stream: Configuration file path or file / file-like object
    :param ac_parser: Forced parser type or parser object
    :param ac_template: Assume configuration file may be a template file and
        try to compile it AAR if True
    :param ac_context: A dict presents context to instantiate template
    :param ac_schema: JSON schema file path to validate given config file
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs: Optional keyword arguments for backends:

        - Common backend options:

          - ignore_missing: Ignore and just return empty result if given file
            (``path_or_stream``) does not exist.

        - Backend specific options such as {"indent": 2} for JSON backend

    :return: Dict-like object (instance of
        anyconfig.mergeabledict.MergeableDict by default) supports merge
        operations.
    """
    is_path_ = is_path(path_or_stream)
    if is_path_:
        path_or_stream = anyconfig.utils.ensure_expandusr(path_or_stream)
        filepath = path_or_stream
    else:
        filepath = anyconfig.utils.get_path_from_stream(path_or_stream)

    psr = find_loader(path_or_stream, ac_parser, is_path_)
    if psr is None:
        return None

    schema = format_checker = None
    if ac_schema is not None:
        kwargs["ac_schema"] = None  # Avoid infinit loop
        format_checker = kwargs.get("format_checker", None)
        LOGGER.info("Loading schema: %s", ac_schema)
        schema = load(ac_schema, ac_parser=None, ac_template=ac_template,
                      ac_context=ac_context, **kwargs)

    LOGGER.info("Loading: %s", filepath)
    if ac_template and filepath is not None:
        try:
            LOGGER.debug("Compiling: %s", filepath)
            content = anyconfig.template.render(filepath, ac_context)
            cnf = psr.loads(content, **kwargs)
            return _maybe_validated(cnf, schema, format_checker)

        except Exception as exc:
            LOGGER.warning("Failed to compile %s, fallback to no template "
                           "mode, exc=%r", path_or_stream, exc)

    cnf = psr.load(path_or_stream, **kwargs)
    return _maybe_validated(cnf, schema, format_checker, ac_namedtuple)


def multi_load(paths, ac_parser=None, ac_template=False, ac_context=None,
               ac_schema=None, ac_namedtuple=False, **kwargs):
    """
    Load multiple config files.

    The first argument `paths` may be a list of config file paths or
    a glob pattern specifying that. That is, if a.yml, b.yml and c.yml are in
    the dir /etc/foo/conf.d/, the followings give same results::

      multi_load(["/etc/foo/conf.d/a.yml", "/etc/foo/conf.d/b.yml",
                  "/etc/foo/conf.d/c.yml", ])

      multi_load("/etc/foo/conf.d/*.yml")

    :param paths: List of config file paths or a glob pattern to list paths, or
        a list of file/file-like objects
    :param ac_parser: Forced parser type or parser object
    :param ac_template: Assume configuration file may be a template file and
        try to compile it AAR if True
    :param ac_context: A dict presents context to instantiate template
    :param ac_schema: JSON schema file path to validate given config file
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs: Optional keyword arguments:

        - Common options:

          - ac_merge (merge): Specify strategy of how to merge results loaded
            from multiple configuration files. See the doc of mergeabledict
            module for more details of strategies. The default is MS_DICTS.

          - ac_marker (marker): Globbing marker to detect paths patterns.

        - Common backend options:

          - ignore_missing: Ignore and just return empty result if given file
            (``path_or_stream``) does not exist.

        - Backend specific options such as {"indent": 2} for JSON backend

    :return: Dict-like object (instance of
        anyconfig.mergeabledict.MergeableDict by default) supports merge
        operations.
    """
    marker = kwargs.setdefault("ac_marker", kwargs.get("marker", '*'))
    ac_merge = kwargs.setdefault("ac_merge", kwargs.get("merge", MS_DICTS))
    if ac_merge not in MERGE_STRATEGIES:
        raise ValueError("Invalid merge strategy: " + ac_merge)

    schema = format_checker = None
    if ac_schema is not None:
        kwargs["ac_schema"] = None  # Avoid infinit loop
        format_checker = kwargs.get("format_checker", None)
        schema = load(ac_schema, ac_parser=None, ac_template=ac_template,
                      ac_context=ac_context, **kwargs)

    cnf = to_container(ac_context) if ac_context else container()
    same_type = anyconfig.utils.are_same_file_types(paths)

    if is_path(paths) and marker in paths:
        paths = anyconfig.utils.sglob(paths)

    for path in paths:
        # Nested patterns like ['*.yml', '/a/b/c.yml'].
        if is_path(path) and marker in path:
            cups = multi_load(path, ac_parser=ac_parser,
                              ac_template=ac_template, ac_context=cnf,
                              **kwargs)
        else:
            if same_type:
                ac_parser = find_loader(path, ac_parser, is_path(path))
            cups = single_load(path, ac_parser=ac_parser,
                               ac_template=ac_template, ac_context=cnf,
                               **kwargs)

        cnf.update(cups, ac_merge)

    return _maybe_validated(cnf, schema, format_checker, ac_namedtuple)


def load(path_specs, ac_parser=None, ac_template=False, ac_context=None,
         ac_schema=None, ac_namedtuple=False, **kwargs):
    r"""
    Load single or multiple config files or multiple config files specified in
    given paths pattern.

    :param path_specs: Configuration file path or paths or its pattern such as
        r'/a/b/\*.json' or a list of files/file-like objects
    :param ac_parser: Forced parser type or parser object
    :param ac_template: Assume configuration file may be a template file and
        try to compile it AAR if True
    :param ac_context: A dict presents context to instantiate template
    :param ac_schema: JSON schema file path to validate given config file
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs:
        Optional keyword arguments. See also the description of `kwargs` in
        `multi_load` function.

    :return: Dict-like object (instance of
        anyconfig.mergeabledict.MergeableDict by default) supports merge
        operations.
    """
    marker = kwargs.setdefault("ac_marker", kwargs.get("marker", '*'))

    if is_path(path_specs) and marker in path_specs or _is_paths(path_specs):
        return multi_load(path_specs, ac_parser=ac_parser,
                          ac_template=ac_template, ac_context=ac_context,
                          ac_schema=ac_schema, ac_namedtuple=ac_namedtuple,
                          **kwargs)
    else:
        return single_load(path_specs, ac_parser=ac_parser,
                           ac_template=ac_template, ac_context=ac_context,
                           ac_schema=ac_schema, ac_namedtuple=ac_namedtuple,
                           **kwargs)


def loads(content, ac_parser=None, ac_template=False, ac_context=None,
          ac_schema=None, ac_namedtuple=False, **kwargs):
    """
    :param content: Configuration file's content
    :param ac_parser: Forced parser type or parser object
    :param ac_template: Assume configuration file may be a template file and
        try to compile it AAR if True
    :param ac_context: Context dict to instantiate template
    :param ac_schema: JSON schema content to validate given config file
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
        JSON loader/dumper backend

    :return: Dict-like object (instance of
        anyconfig.mergeabledict.MergeableDict by default) supports merge
        operations.
    """
    if ac_parser is None:
        LOGGER.warning("No type or parser was given. Try to parse...")
        return anyconfig.parser.parse(content)

    psr = find_loader(None, ac_parser)
    if psr is None:
        return anyconfig.parser.parse(content)

    schema = None
    if ac_schema is not None:
        kwargs["ac_schema"] = None  # Avoid infinit loop
        schema = loads(ac_schema, psr, ac_template, ac_context, **kwargs)

    if ac_template:
        try:
            LOGGER.debug("Compiling")
            content = anyconfig.template.render_s(content, ac_context)
        except Exception as exc:
            LOGGER.warning("Failed to compile and fallback to no template "
                           "mode: '%s', exc=%r", content[:50] + '...', exc)

    cnf = psr.loads(content, **kwargs)
    return _maybe_validated(cnf, schema, kwargs.get("format_checker", None),
                            ac_namedtuple)


def _find_dumper(path_or_stream, ac_parser=None):
    """
    Find configuration parser to dump data.

    :param path_or_stream: Output file path or file / file-like object
    :param ac_parser: Forced parser type or parser object

    :return: Parser-inherited class object
    """
    psr = find_loader(path_or_stream, ac_parser)

    if psr is None or not getattr(psr, "dump", False):
        LOGGER.warning("Dump method not implemented. Fallback to json.Parser")
        psr = anyconfig.backend.json.Parser()

    return psr


def dump(data, path_or_stream, ac_parser=None, ac_namedtuple=False, **kwargs):
    """
    Save `data` as `path_or_stream`.

    :param data: Config data object to dump ::
        anyconfig.mergeabledict.MergeableDict by default
    :param path_or_stream: Output file path or file / file-like object
    :param ac_parser: Forced parser type or parser object
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
        JSON loader/dumper backend
    """
    dumper = _find_dumper(path_or_stream, ac_parser)
    LOGGER.info("Dumping: %s",
                anyconfig.utils.get_path_from_stream(path_or_stream))
    if ac_namedtuple:
        data = create_from(data)
    dumper.dump(data, path_or_stream, **kwargs)


def dumps(data, ac_parser=None, ac_namedtuple=False, **kwargs):
    """
    Return string representation of `data` in forced type format.

    :param data: Config data object to dump ::
        anyconfig.mergeabledict.MergeableDict by default
    :param ac_parser: Forced parser type or parser object
    :param ac_namedtuple: Convert result to nested namedtuple object if True
    :param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
        JSON loader/dumper backend

    :return: Backend-specific string representation for the given data
    """
    if ac_namedtuple:
        data = create_from(data)
    return _find_dumper(None, ac_parser).dumps(data, **kwargs)

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


1			#
2			# Copyright (C) 2012 - 2015 Satoru SATOH <ssato @ redhat.com>
3			# License: MIT
4			#
5			# pylint: disable=unused-import,import-error,invalid-name
6			r"""Public APIs of anyconfig module.
7
8			.. versionadded:: 0.4.99
9			Added ac_namedtuple parameter to \load and \dump APIs.
10
11			.. versionchanged:: 0.3
12			Replaced `forced_type` optional argument of some public APIs with
13			`ac_parser` to allow skip of config parser search by passing parser object
14			previously found and instantiated.
15
16			Also removed some optional arguments, `ignore_missing`, `merge` and
17			`marker`, from definitions of some public APIs as these may not be changed
18			from default in common use cases.
19
20			.. versionchanged:: 0.2
21			Now APIs :func:`find_loader`, :func:`single_load`, :func:`multi_load`,
22			:func:`load` and :func:`dump` can process a file/file-like object or a list
23			of file/file-like objects instead of a file path or a list of file paths.
24
25			.. versionadded:: 0.2
26			Export factory method (create) of anyconfig.mergeabledict.MergeableDict
27			"""
28			from __future__ import absolute_import
29
30			import os.path
31
32			from anyconfig.globals import LOGGER
33			import anyconfig.backends
34			import anyconfig.backend.json
35			import anyconfig.compat
36			import anyconfig.mergeabledict
37			import anyconfig.parser
38			import anyconfig.template
39			import anyconfig.utils
40
41			# Import some global constants will be re-exported:
42			from anyconfig.mergeabledict import (
43			MS_REPLACE, MS_NO_REPLACE, MS_DICTS, MS_DICTS_AND_LISTS, MERGE_STRATEGIES,
44			PATH_SEPS, get, set_, convert_to, create_from # flake8: noqa
45			)
46			from anyconfig.schema import validate, gen_schema
47			from anyconfig.utils import is_path
48
49			# Re-export and aliases:
50			list_types = anyconfig.backends.list_types # flake8: noqa
51			container = anyconfig.mergeabledict.MergeableDict
52			to_container = container.create
53
54
55			def _is_paths(maybe_paths):
56			"""
57			Does given object `maybe_paths` consist of path or path pattern strings?
58			"""
59			if anyconfig.utils.is_iterable(maybe_paths):
60			return not getattr(maybe_paths, "read", False)
61
62			return False # Not an iterable at least.
63
64
65			def _maybe_validated(cnf, schema, format_checker=None, ac_namedtuple=False):
66			"""
67			:param cnf: Configuration object :: container
68			:param schema: JSON schema object :: container
69			:param format_checker: A format property checker object of which class is
70			inherited from jsonschema.FormatChecker, it's default if None given.
71			:param ac_namedtuple: Convert result to nested namedtuple object if True
72
73			:return: Given `cnf` as it is if validation succeeds else None
74			"""
75			if schema is None:
76			valid = True
77			else:
78			(valid, msg) = validate(cnf, schema, format_checker, True)
79			if msg:
80			LOGGER.warning(msg)
81
82			if valid:
83			return convert_to(cnf, True) if ac_namedtuple else cnf
84
85			return None
86
87
88			def set_loglevel(level):
89			"""
90			:param level: Log level, e.g. logging.INFO and logging.WARN.
91			"""
92			LOGGER.setLevel(level)
93
94
95			def find_loader(path_or_stream, parser_or_type=None, is_path_=None):
96			"""
97			Find out config parser object appropriate to load from a file of given path
98			or file/file-like object.
99
100			:param path_or_stream: Configuration file path or file / file-like object
101			:param parser_or_type: Forced configuration parser type or parser object
102			:param is_path_: True if given `path_or_stream` is a file path
103
104			:return: Config parser instance or None
105			"""
106			if anyconfig.backends.is_parser(parser_or_type):
107			return parser_or_type
108
109			(psr, err) = anyconfig.backends.find_parser(path_or_stream, parser_or_type,
110			is_path_=is_path_)
111			if psr is None:
112			LOGGER.error(err)
113			return None
114
115			LOGGER.debug("Using config parser: %r [%s]", psr, psr.type())
116			return psr() # TBD: Passing initialization arguments.
117
118
119			def single_load(path_or_stream, ac_parser=None, ac_template=False,
120			ac_context=None, ac_schema=None, ac_namedtuple=False,
121			**kwargs):
122			"""
123			Load single config file.
124
125			:param path_or_stream: Configuration file path or file / file-like object
126			:param ac_parser: Forced parser type or parser object
127			:param ac_template: Assume configuration file may be a template file and
128			try to compile it AAR if True
129			:param ac_context: A dict presents context to instantiate template
130			:param ac_schema: JSON schema file path to validate given config file
131			:param ac_namedtuple: Convert result to nested namedtuple object if True
132			:param kwargs: Optional keyword arguments for backends:
133
134			- Common backend options:
135
136			- ignore_missing: Ignore and just return empty result if given file
137			(``path_or_stream``) does not exist.
138
139			- Backend specific options such as {"indent": 2} for JSON backend
140
141			:return: Dict-like object (instance of
142			anyconfig.mergeabledict.MergeableDict by default) supports merge
143			operations.
144			"""
145			is_path_ = is_path(path_or_stream)
146			if is_path_:
147			path_or_stream = anyconfig.utils.ensure_expandusr(path_or_stream)
148			filepath = path_or_stream
149			else:
150			filepath = anyconfig.utils.get_path_from_stream(path_or_stream)
151
152			psr = find_loader(path_or_stream, ac_parser, is_path_)
153			if psr is None:
154			return None
155
156			schema = format_checker = None
157			if ac_schema is not None:
158			kwargs["ac_schema"] = None # Avoid infinit loop
159			format_checker = kwargs.get("format_checker", None)
160			LOGGER.info("Loading schema: %s", ac_schema)
161			schema = load(ac_schema, ac_parser=None, ac_template=ac_template,
162			ac_context=ac_context, **kwargs)
163
164			LOGGER.info("Loading: %s", filepath)
165			if ac_template and filepath is not None:
166			try:
167			LOGGER.debug("Compiling: %s", filepath)
168			content = anyconfig.template.render(filepath, ac_context)
169			cnf = psr.loads(content, **kwargs)
170			return _maybe_validated(cnf, schema, format_checker)
171
172			except Exception as exc:
173			LOGGER.warning("Failed to compile %s, fallback to no template "
174			"mode, exc=%r", path_or_stream, exc)
175
176			cnf = psr.load(path_or_stream, **kwargs)
177			return _maybe_validated(cnf, schema, format_checker, ac_namedtuple)
178
179
180			def multi_load(paths, ac_parser=None, ac_template=False, ac_context=None,
181			ac_schema=None, ac_namedtuple=False, **kwargs):
182			"""
183			Load multiple config files.
184
185			The first argument `paths` may be a list of config file paths or
186			a glob pattern specifying that. That is, if a.yml, b.yml and c.yml are in
187			the dir /etc/foo/conf.d/, the followings give same results::
188
189			multi_load(["/etc/foo/conf.d/a.yml", "/etc/foo/conf.d/b.yml",
190			"/etc/foo/conf.d/c.yml", ])
191
192			multi_load("/etc/foo/conf.d/*.yml")
193
194			:param paths: List of config file paths or a glob pattern to list paths, or
195			a list of file/file-like objects
196			:param ac_parser: Forced parser type or parser object
197			:param ac_template: Assume configuration file may be a template file and
198			try to compile it AAR if True
199			:param ac_context: A dict presents context to instantiate template
200			:param ac_schema: JSON schema file path to validate given config file
201			:param ac_namedtuple: Convert result to nested namedtuple object if True
202			:param kwargs: Optional keyword arguments:
203
204			- Common options:
205
206			- ac_merge (merge): Specify strategy of how to merge results loaded
207			from multiple configuration files. See the doc of mergeabledict
208			module for more details of strategies. The default is MS_DICTS.
209
210			- ac_marker (marker): Globbing marker to detect paths patterns.
211
212			- Common backend options:
213
214			- ignore_missing: Ignore and just return empty result if given file
215			(``path_or_stream``) does not exist.
216
217			- Backend specific options such as {"indent": 2} for JSON backend
218
219			:return: Dict-like object (instance of
220			anyconfig.mergeabledict.MergeableDict by default) supports merge
221			operations.
222			"""
223			marker = kwargs.setdefault("ac_marker", kwargs.get("marker", '*'))
224			ac_merge = kwargs.setdefault("ac_merge", kwargs.get("merge", MS_DICTS))
225			if ac_merge not in MERGE_STRATEGIES:
226			raise ValueError("Invalid merge strategy: " + ac_merge)
227
228			schema = format_checker = None
229			if ac_schema is not None:
230			kwargs["ac_schema"] = None # Avoid infinit loop
231			format_checker = kwargs.get("format_checker", None)
232			schema = load(ac_schema, ac_parser=None, ac_template=ac_template,
233			ac_context=ac_context, **kwargs)
234
235			cnf = to_container(ac_context) if ac_context else container()
236			same_type = anyconfig.utils.are_same_file_types(paths)
237
238			if is_path(paths) and marker in paths:
239			paths = anyconfig.utils.sglob(paths)
240
241			for path in paths:
242			# Nested patterns like ['*.yml', '/a/b/c.yml'].
243			if is_path(path) and marker in path:
244			cups = multi_load(path, ac_parser=ac_parser,
245			ac_template=ac_template, ac_context=cnf,
246			**kwargs)
247			else:
248			if same_type:
249			ac_parser = find_loader(path, ac_parser, is_path(path))
250			cups = single_load(path, ac_parser=ac_parser,
251			ac_template=ac_template, ac_context=cnf,
252			**kwargs)
253
254			cnf.update(cups, ac_merge)
255
256			return _maybe_validated(cnf, schema, format_checker, ac_namedtuple)
257
258
259			def load(path_specs, ac_parser=None, ac_template=False, ac_context=None,
260			ac_schema=None, ac_namedtuple=False, **kwargs):
261			r"""
262			Load single or multiple config files or multiple config files specified in
263			given paths pattern.
264
265			:param path_specs: Configuration file path or paths or its pattern such as
266			r'/a/b/\*.json' or a list of files/file-like objects
267			:param ac_parser: Forced parser type or parser object
268			:param ac_template: Assume configuration file may be a template file and
269			try to compile it AAR if True
270			:param ac_context: A dict presents context to instantiate template
271			:param ac_schema: JSON schema file path to validate given config file
272			:param ac_namedtuple: Convert result to nested namedtuple object if True
273			:param kwargs:
274			Optional keyword arguments. See also the description of `kwargs` in
275			`multi_load` function.
276
277			:return: Dict-like object (instance of
278			anyconfig.mergeabledict.MergeableDict by default) supports merge
279			operations.
280			"""
281			marker = kwargs.setdefault("ac_marker", kwargs.get("marker", '*'))
282
283			if is_path(path_specs) and marker in path_specs or _is_paths(path_specs):
284			return multi_load(path_specs, ac_parser=ac_parser,
285			ac_template=ac_template, ac_context=ac_context,
286			ac_schema=ac_schema, ac_namedtuple=ac_namedtuple,
287			**kwargs)
288			else:
289			return single_load(path_specs, ac_parser=ac_parser,
290			ac_template=ac_template, ac_context=ac_context,
291			ac_schema=ac_schema, ac_namedtuple=ac_namedtuple,
292			**kwargs)
293
294
295			def loads(content, ac_parser=None, ac_template=False, ac_context=None,
296			ac_schema=None, ac_namedtuple=False, **kwargs):
297			"""
298			:param content: Configuration file's content
299			:param ac_parser: Forced parser type or parser object
300			:param ac_template: Assume configuration file may be a template file and
301			try to compile it AAR if True
302			:param ac_context: Context dict to instantiate template
303			:param ac_schema: JSON schema content to validate given config file
304			:param ac_namedtuple: Convert result to nested namedtuple object if True
305			:param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
306			JSON loader/dumper backend
307
308			:return: Dict-like object (instance of
309			anyconfig.mergeabledict.MergeableDict by default) supports merge
310			operations.
311			"""
312			if ac_parser is None:
313			LOGGER.warning("No type or parser was given. Try to parse...")
314			return anyconfig.parser.parse(content)
315
316			psr = find_loader(None, ac_parser)
317			if psr is None:
318			return anyconfig.parser.parse(content)
319
320			schema = None
321			if ac_schema is not None:
322			kwargs["ac_schema"] = None # Avoid infinit loop
323			schema = loads(ac_schema, psr, ac_template, ac_context, **kwargs)
324
325			if ac_template:
326			try:
327			LOGGER.debug("Compiling")
328			content = anyconfig.template.render_s(content, ac_context)
329			except Exception as exc:
330			LOGGER.warning("Failed to compile and fallback to no template "
331			"mode: '%s', exc=%r", content[:50] + '...', exc)
332
333			cnf = psr.loads(content, **kwargs)
334			return _maybe_validated(cnf, schema, kwargs.get("format_checker", None),
335			ac_namedtuple)
336
337
338			def _find_dumper(path_or_stream, ac_parser=None):
339			"""
340			Find configuration parser to dump data.
341
342			:param path_or_stream: Output file path or file / file-like object
343			:param ac_parser: Forced parser type or parser object
344
345			:return: Parser-inherited class object
346			"""
347			psr = find_loader(path_or_stream, ac_parser)
348
349			if psr is None or not getattr(psr, "dump", False):
350			LOGGER.warning("Dump method not implemented. Fallback to json.Parser")
351			psr = anyconfig.backend.json.Parser()
352
353			return psr
354
355
356			def dump(data, path_or_stream, ac_parser=None, ac_namedtuple=False, **kwargs):
357			"""
358			Save `data` as `path_or_stream`.
359
360			:param data: Config data object to dump ::
361			anyconfig.mergeabledict.MergeableDict by default
362			:param path_or_stream: Output file path or file / file-like object
363			:param ac_parser: Forced parser type or parser object
364			:param ac_namedtuple: Convert result to nested namedtuple object if True
365			:param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
366			JSON loader/dumper backend
367			"""
368			dumper = _find_dumper(path_or_stream, ac_parser)
369			LOGGER.info("Dumping: %s",
370			anyconfig.utils.get_path_from_stream(path_or_stream))
371			if ac_namedtuple:
372			data = create_from(data)
373			dumper.dump(data, path_or_stream, **kwargs)
374
375
376			def dumps(data, ac_parser=None, ac_namedtuple=False, **kwargs):
377			"""
378			Return string representation of `data` in forced type format.
379
380			:param data: Config data object to dump ::
381			anyconfig.mergeabledict.MergeableDict by default
382			:param ac_parser: Forced parser type or parser object
383			:param ac_namedtuple: Convert result to nested namedtuple object if True
384			:param kwargs: Backend specific optional arguments, e.g. {"indent": 2} for
385			JSON loader/dumper backend
386
387			:return: Backend-specific string representation for the given data
388			"""
389			if ac_namedtuple:
390			data = create_from(data)
391			return _find_dumper(None, ac_parser).dumps(data, **kwargs)
392
393			# vim:sw=4:ts=4:et:
394

ssato / python-anyconfig

Push — master ( c877c2...3e5906 )

anyconfig._maybe_validated() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like