anyconfig.single_load() - Code Metrics - Inspection of "api: consolidate duplicated schema processing code..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 34d22d...8ebabd )

by Satoru

created 2015-12-18 01:52 UTC

anyconfig.single_load() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	6
dl	0
loc	57
rs	7.8388

How to fix Long Method

#
# 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
   Most keyword arguments passed to APIs are now position independent.

.. 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, **options):
    """
    :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 options: Keyword options such as:

        - 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=format_checker,
                                safe=True)
        if msg:
            LOGGER.warning(msg)

    if valid:
        return convert_to(cnf, True) if options.get("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 _load_schema(**options):
    """
    :param options: Optional keyword arguments such as

        - ac_template: Assume configuration file may be a template file and try
          to compile it AAR if True
        - ac_context: A dict presents context to instantiate template
        - ac_schema: JSON schema file path to validate given config file
    """
    ac_schema = options.get("ac_schema", None)
    if ac_schema is not None:
        options["ac_parser"] = None  # Try to detect it as it may be different
                                     # format from the original config file.
        options["ac_schema"] = None  # Avoid infinite loop.
        LOGGER.info("Loading schema: %s", ac_schema)
        return load(ac_schema, **options)

    return None


def single_load(path_or_stream, ac_parser=None, ac_template=False,
                ac_context=None, **options):
    """
    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 options: Optional keyword arguments such as:

        - Common options:

          - ac_namedtuple: Convert result to nested namedtuple object if True
          - ac_schema: JSON schema file path to validate given config file

        - 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 = _load_schema(ac_template=ac_template, ac_context=ac_context,
                          **options)
    options["ac_schema"] = None  # It's not needed now.

    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, **options)
            return _maybe_validated(cnf, schema, **options)

        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, **options)
    return _maybe_validated(cnf, schema, **options)


def multi_load(paths, ac_parser=None, ac_template=False, ac_context=None,
               **options):
    """
    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 options: 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.
          - ac_namedtuple: Convert result to nested namedtuple object if True
          - ac_schema: JSON schema file path to validate given config file

        - 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 = options.setdefault("ac_marker", options.get("marker", '*'))
    ac_merge = options.setdefault("ac_merge", options.get("merge", MS_DICTS))
    if ac_merge not in MERGE_STRATEGIES:
        raise ValueError("Invalid merge strategy: " + ac_merge)

    schema = _load_schema(ac_template=ac_template, ac_context=ac_context,
                          **options)
    options["ac_schema"] = None  # It's not needed now.

    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:
        opts = options.copy()
        opts["ac_namedtuple"] = False  # Disabled within this loop.
        # 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, **opts)
        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, **opts)

        cnf.update(cups, ac_merge)

    return _maybe_validated(cnf, schema, **options)


def load(path_specs, ac_parser=None, ac_template=False, ac_context=None,
         **options):
    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 options:
        Optional keyword arguments. See also the description of `options` in
        `multi_load` function.

    :return: Dict-like object (instance of
        anyconfig.mergeabledict.MergeableDict by default) supports merge
        operations.
    """
    marker = options.setdefault("ac_marker", options.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,
                          **options)
    else:
        return single_load(path_specs, ac_parser=ac_parser,
                           ac_template=ac_template, ac_context=ac_context,
                           **options)


def loads(content, ac_parser=None, ac_template=False, ac_context=None,
          **options):
    """
    :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 options:
        Optional keyword arguments. See also the description of `options` in
        `single_load` function.

    :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
    ac_schema = options.get("ac_schema", None)
    if ac_schema is not None:
        options["ac_schema"] = None
        schema = loads(ac_schema, ac_parser=psr, ac_template=ac_template,
                       ac_context=ac_context, **options)

    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, **options)
    return _maybe_validated(cnf, schema, **options)


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, **options):
    """
    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 options: Keyword options such ash:

        - ac_namedtuple: Convert result to nested namedtuple object if True
        - other 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 options.get("ac_namedtuple", False):
        data = create_from(data)
    dumper.dump(data, path_or_stream, **options)


def dumps(data, ac_parser=None, **options):
    """
    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 options: see :func:`dump`

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

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

ssato / python-anyconfig

Push — master ( 34d22d...8ebabd )

anyconfig.single_load() B

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like