_do_get() - Code Metrics - Inspection of "refactor: consolidate filter (query, get and set)..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( dde88c...c42960 )

by Satoru

created 2017-03-11 07:59 UTC

_do_get() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	2
c	1
b	0
f	0
dl	0
loc	11
rs	9.4285

#
# Author: Satoru SATOH <ssato redhat.com>
# License: MIT
#
"""CLI frontend module for anyconfig.
"""
from __future__ import absolute_import, print_function

import argparse
import codecs
import locale
import logging
import os
import sys

import anyconfig.api as API
import anyconfig.compat
import anyconfig.globals
import anyconfig.parser
import anyconfig.utils


_ENCODING = locale.getdefaultlocale()[1]

LOGGER = logging.getLogger("anyconfig")
LOGGER.addHandler(logging.StreamHandler())
LOGGER.setLevel(logging.WARN)

if anyconfig.compat.IS_PYTHON_3:
    import io

    _ENCODING = _ENCODING.lower()

    # TODO: What should be done for an error, "AttributeError: '_io.StringIO'
    # object has no attribute 'buffer'"?
    try:
        sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding=_ENCODING)
        sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding=_ENCODING)
    except AttributeError:
        pass
else:
    sys.stdout = codecs.getwriter(_ENCODING)(sys.stdout)
    sys.stderr = codecs.getwriter(_ENCODING)(sys.stderr)

USAGE = """\
%(prog)s [Options...] CONF_PATH_OR_PATTERN_0 [CONF_PATH_OR_PATTERN_1 ..]

Examples:
  %(prog)s --list  # -> Supported config types: configobj, ini, json, ...
  # Merge and/or convert input config to output config [file]
  %(prog)s -I yaml -O yaml /etc/xyz/conf.d/a.conf
  %(prog)s -I yaml '/etc/xyz/conf.d/*.conf' -o xyz.conf --otype json
  %(prog)s '/etc/xyz/conf.d/*.json' -o xyz.yml \\
    --atype json -A '{"obsoletes": "syscnf", "conflicts": "syscnf-old"}'
  %(prog)s '/etc/xyz/conf.d/*.json' -o xyz.yml \\
    -A obsoletes:syscnf;conflicts:syscnf-old
  %(prog)s /etc/foo.json /etc/foo/conf.d/x.json /etc/foo/conf.d/y.json
  %(prog)s '/etc/foo.d/*.json' -M noreplace
  # Query/Get/set part of input config
  %(prog)s '/etc/foo.d/*.json' --query 'locs[?state == 'T'].name | sort(@)'
  %(prog)s '/etc/foo.d/*.json' --get a.b.c
  %(prog)s '/etc/foo.d/*.json' --set a.b.c=1
  # Validate with JSON schema or generate JSON schema:
  %(prog)s --validate -S foo.conf.schema.yml '/etc/foo.d/*.xml'
  %(prog)s --gen-schema '/etc/foo.d/*.xml' -o foo.conf.schema.yml"""

DEFAULTS = dict(loglevel=1, list=False, output=None, itype=None,
                otype=None, atype=None, merge=API.MS_DICTS,
                ignore_missing=False, template=False, env=False,
                schema=None, validate=False, gen_schema=False)


def to_log_level(level):
    """
    :param level: Logging level in int = 0 .. 2

    >>> to_log_level(0) == logging.WARN
    True
    >>> to_log_level(5)  # doctest: +IGNORE_EXCEPTION_DETAIL, +ELLIPSIS
    Traceback (most recent call last):
        ...
    ValueError: wrong log level passed: 5
    >>>
    """
    if not (level >= 0 and level < 3):
        raise ValueError("wrong log level passed: " + str(level))

    return [logging.WARN, logging.INFO, logging.DEBUG][level]


_ATYPE_HELP_FMT = """\
Explicitly select type of argument to provide configs from %s.

If this option is not set, original parser is used: 'K:V' will become {K: V},
'K:V_0,V_1,..' will become {K: [V_0, V_1, ...]}, and 'K_0:V_0;K_1:V_1' will
become {K_0: V_0, K_1: V_1} (where the tyep of K is str, type of V is one of
Int, str, etc."""

_QUERY_HELP = ("Query with JMESPath expression language. See "
               "http://jmespath.org for more about JMESPath expression. "
               "This option is not used with --get option at the same time. "
               "Please note that python module to support JMESPath "
               "expression (https://pypi.python.org/pypi/jmespath/) is "
               "required to use this option")
_GET_HELP = ("Specify key path to get part of config, for example, "
             "'--get a.b.c' to config {'a': {'b': {'c': 0, 'd': 1}}} "
             "gives 0 and '--get a.b' to the same config gives "
             "{'c': 0, 'd': 1}. Path expression can be JSON Pointer "
             "expression (http://tools.ietf.org/html/rfc6901) such like "
             "'', '/a~1b', '/m~0n'. "
             "This option is not used with --query option at the same time. ")
_SET_HELP = ("Specify key path to set (update) part of config, for "
             "example, '--set a.b.c=1' to a config {'a': {'b': {'c': 0, "
             "'d': 1}}} gives {'a': {'b': {'c': 1, 'd': 1}}}.")


def make_parser(defaults=None):
    """
    :param defaults: Default option values
    """
    if defaults is None:
        defaults = DEFAULTS

    ctypes = API.list_types()
    ctypes_s = ", ".join(ctypes)
    type_help = "Select type of %s config files from " + \
        ctypes_s + " [Automatically detected by file ext]"

    mts = API.MERGE_STRATEGIES
    mts_s = ", ".join(mts)
    mt_help = "Select strategy to merge multiple configs from " + \
        mts_s + " [%(merge)s]" % defaults

    parser = argparse.ArgumentParser(USAGE)
    parser.set_defaults(**defaults)

    parser.add_argument("inputs", type=str, nargs='*', help="Input files")
    parser.add_argument("--version", action="version",
                        version="%%(prog)s %s" % anyconfig.globals.VERSION)

    lpog = parser.add_argument_group("List specific options")
    lpog.add_argument("-L", "--list", action="store_true",
                      help="List supported config types")

    spog = parser.add_argument_group("Schema specific options")
    spog.add_argument("--validate", action="store_true",
                      help="Only validate input files and do not output. "
                           "You must specify schema file with -S/--schema "
                           "option.")
    spog.add_argument("--gen-schema", action="store_true",
                      help="Generate JSON schema for givne config file[s] "
                           "and output it instead of (merged) configuration.")

    gspog = parser.add_argument_group("Query/Get/set options")
    gspog.add_argument("-Q", "--query", help=_QUERY_HELP)
    gspog.add_argument("--get", help=_GET_HELP)
    gspog.add_argument("--set", help=_SET_HELP)

    parser.add_argument("-o", "--output", help="Output file path")
    parser.add_argument("-I", "--itype", choices=ctypes,
                        help=(type_help % "Input"))
    parser.add_argument("-O", "--otype", choices=ctypes,
                        help=(type_help % "Output"))
    parser.add_argument("-M", "--merge", choices=mts, help=mt_help)
    parser.add_argument("-A", "--args", help="Argument configs to override")
    parser.add_argument("--atype", choices=ctypes,
                        help=_ATYPE_HELP_FMT % ctypes_s)

    parser.add_argument("-x", "--ignore-missing", action="store_true",
                        help="Ignore missing input files")
    parser.add_argument("-T", "--template", action="store_true",
                        help="Enable template config support")
    parser.add_argument("-E", "--env", action="store_true",
                        help="Load configuration defaults from "
                             "environment values")
    parser.add_argument("-S", "--schema", help="Specify Schema file[s] path")
    parser.add_argument("-s", "--silent", action="store_const",
                        dest="loglevel", const=0, help="Silent or quiet mode")
    parser.add_argument("-q", "--quiet", action="store_const", dest="loglevel",
                        const=0, help="Same as --silent option")
    parser.add_argument("-v", "--verbose", action="store_const",
                        dest="loglevel", const=2, help="Verbose mode")
    return parser


def _exit_with_output(content, exit_code=0):
    """
    Exit the program with printing out messages.

    :param content: content to print out
    :param exit_code: Exit code
    """
    (sys.stdout if exit_code == 0 else sys.stderr).write(content + "\n")
    sys.exit(exit_code)


def _parse_args(argv):
    """
    Show supported config format types or usage.

    :param argv: Argument list to parse or None (sys.argv will be set).
    :return: argparse.Namespace object or None (exit before return)
    """
    parser = make_parser()
    args = parser.parse_args(argv)
    LOGGER.setLevel(to_log_level(args.loglevel))

    if not args.inputs:
        if args.list:
            tlist = ", ".join(API.list_types())
            _exit_with_output("Supported config types: " + tlist)
        elif args.env:
            cnf = os.environ.copy()
            _output_result(cnf, args.output, args.otype or "json", None, None)
            sys.exit(0)
        else:
            parser.print_usage()
            sys.exit(1)

    if args.validate and args.schema is None:
        _exit_with_output("--validate option requires --scheme option", 1)

    return args


def _exit_if_load_failure(cnf, msg):
    """
    :param cnf: Loaded configuration object or None indicates load failure
    :param msg: Message to print out if failure
    """
    if cnf is None:
        _exit_with_output(msg, 1)


def _do_get(cnf, get_path):
    """
    :param cnf: Configuration object to print out
    :param get_path: key path given in --get option
    :return: updated Configuration object if no error
    """
    (cnf, err) = API.get(cnf, get_path)
    if cnf is None:  # Failed to get the result.
        _exit_with_output("Failed to get result: err=%s" % err, 1)

    return cnf


def _output_type_by_input_path(inpath, itype, fmsg):
    """
    :param inpath: Input file path
    :param itype: Input type or None
    :param fmsg: message if it cannot detect otype by `inpath`
    :return: Output type :: str
    """
    msg = ("Specify inpath and/or outpath type[s] with -I/--itype "
           "or -O/--otype option explicitly")
    if itype is None:
        try:
            otype = API.find_loader(inpath).type()
        except API.UnknownFileTypeError:
            _exit_with_output((fmsg % inpath) + msg, 1)
        except ValueError:
            _exit_with_output(msg, 1)
    else:
        otype = itype

    return otype


def _try_dump(cnf, outpath, otype, fmsg):
    """
    :param cnf: Configuration object to print out
    :param outpath: Output file path or None
    :param otype: Output type or None
    :param fmsg: message if it cannot detect otype by `inpath`
    """
    try:
        API.dump(cnf, outpath, otype)
    except API.UnknownFileTypeError:
        _exit_with_output(fmsg % outpath, 1)
    except API.UnknownParserTypeError:
        _exit_with_output("Invalid output type '%s'" % otype, 1)


def _output_result(cnf, outpath, otype, inpath, itype):
    """
    :param cnf: Configuration object to print out
    :param outpath: Output file path or None
    :param otype: Output type or None
    :param inpath: Input file path
    :param itype: Input type or None
    """
    fmsg = ("Uknown file type and cannot detect appropriate backend "
            "from its extension, '%s'")

    if not anyconfig.utils.is_dict_like(cnf):
        _exit_with_output(str(cnf))  # Print primitive types as it is.

    if not outpath or outpath == "-":
        outpath = sys.stdout
        if otype is None:
            otype = _output_type_by_input_path(inpath, itype, fmsg)

    _try_dump(cnf, outpath, otype, fmsg)


def _load_diff(args):
    """
    :param args: :class:`~argparse.Namespace` object
    """
    try:
        diff = API.load(args.inputs, args.itype,
                        ignore_missing=args.ignore_missing,
                        ac_merge=args.merge,
                        ac_template=args.template,
                        ac_schema=args.schema)
    except API.UnknownParserTypeError:
        _exit_with_output("Wrong input type '%s'" % args.itype, 1)
    except API.UnknownFileTypeError:
        _exit_with_output("No appropriate backend was found for given file "
                          "'%s'" % args.itype, 1)
    _exit_if_load_failure(diff,
                          "Failed to load: args=%s" % ", ".join(args.inputs))

    return diff


def _do_filter(cnf, args):
    """
    :param cnf: Mapping object represents configuration data
    :param args: :class:`~argparse.Namespace` object
    :return: `cnf` may be updated
    """
    if args.query:
        cnf = API.query(cnf, args.query)
    elif args.get:
        cnf = _do_get(cnf, args.get)
    elif args.set:
        (key, val) = args.set.split('=')
        API.set_(cnf, key, anyconfig.parser.parse(val))

    return cnf


def main(argv=None):
    """
    :param argv: Argument list to parse or None (sys.argv will be set).
    """
    args = _parse_args((argv if argv else sys.argv)[1:])
    cnf = os.environ.copy() if args.env else {}
    diff = _load_diff(args)
    API.merge(cnf, diff)

    if args.args:
        diff = anyconfig.parser.parse(args.args)
        API.merge(cnf, diff)

    if args.validate:
        _exit_with_output("Validation succeds")

    cnf = API.gen_schema(cnf) if args.gen_schema else _do_filter(cnf, args)

    _output_result(cnf, args.output, args.otype, args.inputs, args.itype)


if __name__ == '__main__':
    main(sys.argv)

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


1			#
2			# Author: Satoru SATOH <ssato redhat.com>
3			# License: MIT
4			#
5			"""CLI frontend module for anyconfig.
6			"""
7			from __future__ import absolute_import, print_function
8
9			import argparse
10			import codecs
11			import locale
12			import logging
13			import os
14			import sys
15
16			import anyconfig.api as API
17			import anyconfig.compat
18			import anyconfig.globals
19			import anyconfig.parser
20			import anyconfig.utils
21
22
23			_ENCODING = locale.getdefaultlocale()[1]
24
25			LOGGER = logging.getLogger("anyconfig")
26			LOGGER.addHandler(logging.StreamHandler())
27			LOGGER.setLevel(logging.WARN)
28
29			if anyconfig.compat.IS_PYTHON_3:
30			import io
31
32			_ENCODING = _ENCODING.lower()
33
34			# TODO: What should be done for an error, "AttributeError: '_io.StringIO'
35			# object has no attribute 'buffer'"?
36			try:
37			sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding=_ENCODING)
38			sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding=_ENCODING)
39			except AttributeError:
40			pass
41			else:
42			sys.stdout = codecs.getwriter(_ENCODING)(sys.stdout)
43			sys.stderr = codecs.getwriter(_ENCODING)(sys.stderr)
44
45			USAGE = """\
46			%(prog)s [Options...] CONF_PATH_OR_PATTERN_0 [CONF_PATH_OR_PATTERN_1 ..]
47
48			Examples:
49			%(prog)s --list # -> Supported config types: configobj, ini, json, ...
50			# Merge and/or convert input config to output config [file]
51			%(prog)s -I yaml -O yaml /etc/xyz/conf.d/a.conf
52			%(prog)s -I yaml '/etc/xyz/conf.d/*.conf' -o xyz.conf --otype json
53			%(prog)s '/etc/xyz/conf.d/*.json' -o xyz.yml \\
54			--atype json -A '{"obsoletes": "syscnf", "conflicts": "syscnf-old"}'
55			%(prog)s '/etc/xyz/conf.d/*.json' -o xyz.yml \\
56			-A obsoletes:syscnf;conflicts:syscnf-old
57			%(prog)s /etc/foo.json /etc/foo/conf.d/x.json /etc/foo/conf.d/y.json
58			%(prog)s '/etc/foo.d/*.json' -M noreplace
59			# Query/Get/set part of input config
60			%(prog)s '/etc/foo.d/*.json' --query 'locs[?state == 'T'].name \| sort(@)'
61			%(prog)s '/etc/foo.d/*.json' --get a.b.c
62			%(prog)s '/etc/foo.d/*.json' --set a.b.c=1
63			# Validate with JSON schema or generate JSON schema:
64			%(prog)s --validate -S foo.conf.schema.yml '/etc/foo.d/*.xml'
65			%(prog)s --gen-schema '/etc/foo.d/*.xml' -o foo.conf.schema.yml"""
66
67			DEFAULTS = dict(loglevel=1, list=False, output=None, itype=None,
68			otype=None, atype=None, merge=API.MS_DICTS,
69			ignore_missing=False, template=False, env=False,
70			schema=None, validate=False, gen_schema=False)
71
72
73			def to_log_level(level):
74			"""
75			:param level: Logging level in int = 0 .. 2
76
77			>>> to_log_level(0) == logging.WARN
78			True
79			>>> to_log_level(5) # doctest: +IGNORE_EXCEPTION_DETAIL, +ELLIPSIS
80			Traceback (most recent call last):
81			...
82			ValueError: wrong log level passed: 5
83			>>>
84			"""
85			if not (level >= 0 and level < 3):
86			raise ValueError("wrong log level passed: " + str(level))
87
88			return [logging.WARN, logging.INFO, logging.DEBUG][level]
89
90
91			_ATYPE_HELP_FMT = """\
92			Explicitly select type of argument to provide configs from %s.
93
94			If this option is not set, original parser is used: 'K:V' will become {K: V},
95			'K:V_0,V_1,..' will become {K: [V_0, V_1, ...]}, and 'K_0:V_0;K_1:V_1' will
96			become {K_0: V_0, K_1: V_1} (where the tyep of K is str, type of V is one of
97			Int, str, etc."""
98
99			_QUERY_HELP = ("Query with JMESPath expression language. See "
100			"http://jmespath.org for more about JMESPath expression. "
101			"This option is not used with --get option at the same time. "
102			"Please note that python module to support JMESPath "
103			"expression (https://pypi.python.org/pypi/jmespath/) is "
104			"required to use this option")
105			_GET_HELP = ("Specify key path to get part of config, for example, "
106			"'--get a.b.c' to config {'a': {'b': {'c': 0, 'd': 1}}} "
107			"gives 0 and '--get a.b' to the same config gives "
108			"{'c': 0, 'd': 1}. Path expression can be JSON Pointer "
109			"expression (http://tools.ietf.org/html/rfc6901) such like "
110			"'', '/a~1b', '/m~0n'. "
111			"This option is not used with --query option at the same time. ")
112			_SET_HELP = ("Specify key path to set (update) part of config, for "
113			"example, '--set a.b.c=1' to a config {'a': {'b': {'c': 0, "
114			"'d': 1}}} gives {'a': {'b': {'c': 1, 'd': 1}}}.")
115
116
117			def make_parser(defaults=None):
118			"""
119			:param defaults: Default option values
120			"""
121			if defaults is None:
122			defaults = DEFAULTS
123
124			ctypes = API.list_types()
125			ctypes_s = ", ".join(ctypes)
126			type_help = "Select type of %s config files from " + \
127			ctypes_s + " [Automatically detected by file ext]"
128
129			mts = API.MERGE_STRATEGIES
130			mts_s = ", ".join(mts)
131			mt_help = "Select strategy to merge multiple configs from " + \
132			mts_s + " [%(merge)s]" % defaults
133
134			parser = argparse.ArgumentParser(USAGE)
135			parser.set_defaults(**defaults)
136
137			parser.add_argument("inputs", type=str, nargs='*', help="Input files")
138			parser.add_argument("--version", action="version",
139			version="%%(prog)s %s" % anyconfig.globals.VERSION)
140
141			lpog = parser.add_argument_group("List specific options")
142			lpog.add_argument("-L", "--list", action="store_true",
143			help="List supported config types")
144
145			spog = parser.add_argument_group("Schema specific options")
146			spog.add_argument("--validate", action="store_true",
147			help="Only validate input files and do not output. "
148			"You must specify schema file with -S/--schema "
149			"option.")
150			spog.add_argument("--gen-schema", action="store_true",
151			help="Generate JSON schema for givne config file[s] "
152			"and output it instead of (merged) configuration.")
153
154			gspog = parser.add_argument_group("Query/Get/set options")
155			gspog.add_argument("-Q", "--query", help=_QUERY_HELP)
156			gspog.add_argument("--get", help=_GET_HELP)
157			gspog.add_argument("--set", help=_SET_HELP)
158
159			parser.add_argument("-o", "--output", help="Output file path")
160			parser.add_argument("-I", "--itype", choices=ctypes,
161			help=(type_help % "Input"))
162			parser.add_argument("-O", "--otype", choices=ctypes,
163			help=(type_help % "Output"))
164			parser.add_argument("-M", "--merge", choices=mts, help=mt_help)
165			parser.add_argument("-A", "--args", help="Argument configs to override")
166			parser.add_argument("--atype", choices=ctypes,
167			help=_ATYPE_HELP_FMT % ctypes_s)
168
169			parser.add_argument("-x", "--ignore-missing", action="store_true",
170			help="Ignore missing input files")
171			parser.add_argument("-T", "--template", action="store_true",
172			help="Enable template config support")
173			parser.add_argument("-E", "--env", action="store_true",
174			help="Load configuration defaults from "
175			"environment values")
176			parser.add_argument("-S", "--schema", help="Specify Schema file[s] path")
177			parser.add_argument("-s", "--silent", action="store_const",
178			dest="loglevel", const=0, help="Silent or quiet mode")
179			parser.add_argument("-q", "--quiet", action="store_const", dest="loglevel",
180			const=0, help="Same as --silent option")
181			parser.add_argument("-v", "--verbose", action="store_const",
182			dest="loglevel", const=2, help="Verbose mode")
183			return parser
184
185
186			def _exit_with_output(content, exit_code=0):
187			"""
188			Exit the program with printing out messages.
189
190			:param content: content to print out
191			:param exit_code: Exit code
192			"""
193			(sys.stdout if exit_code == 0 else sys.stderr).write(content + "\n")
194			sys.exit(exit_code)
195
196
197			def _parse_args(argv):
198			"""
199			Show supported config format types or usage.
200
201			:param argv: Argument list to parse or None (sys.argv will be set).
202			:return: argparse.Namespace object or None (exit before return)
203			"""
204			parser = make_parser()
205			args = parser.parse_args(argv)
206			LOGGER.setLevel(to_log_level(args.loglevel))
207
208			if not args.inputs:
209			if args.list:
210			tlist = ", ".join(API.list_types())
211			_exit_with_output("Supported config types: " + tlist)
212			elif args.env:
213			cnf = os.environ.copy()
214			_output_result(cnf, args.output, args.otype or "json", None, None)
215			sys.exit(0)
216			else:
217			parser.print_usage()
218			sys.exit(1)
219
220			if args.validate and args.schema is None:
221			_exit_with_output("--validate option requires --scheme option", 1)
222
223			return args
224
225
226			def _exit_if_load_failure(cnf, msg):
227			"""
228			:param cnf: Loaded configuration object or None indicates load failure
229			:param msg: Message to print out if failure
230			"""
231			if cnf is None:
232			_exit_with_output(msg, 1)
233
234
235			def _do_get(cnf, get_path):
236			"""
237			:param cnf: Configuration object to print out
238			:param get_path: key path given in --get option
239			:return: updated Configuration object if no error
240			"""
241			(cnf, err) = API.get(cnf, get_path)
242			if cnf is None: # Failed to get the result.
243			_exit_with_output("Failed to get result: err=%s" % err, 1)
244
245			return cnf
246
247
248			def _output_type_by_input_path(inpath, itype, fmsg):
249			"""
250			:param inpath: Input file path
251			:param itype: Input type or None
252			:param fmsg: message if it cannot detect otype by `inpath`
253			:return: Output type :: str
254			"""
255			msg = ("Specify inpath and/or outpath type[s] with -I/--itype "
256			"or -O/--otype option explicitly")
257			if itype is None:
258			try:
259			otype = API.find_loader(inpath).type()
260			except API.UnknownFileTypeError:
261			_exit_with_output((fmsg % inpath) + msg, 1)
262			except ValueError:
263			_exit_with_output(msg, 1)
264			else:
265			otype = itype
266
267			return otype
268
269
270			def _try_dump(cnf, outpath, otype, fmsg):
271			"""
272			:param cnf: Configuration object to print out
273			:param outpath: Output file path or None
274			:param otype: Output type or None
275			:param fmsg: message if it cannot detect otype by `inpath`
276			"""
277			try:
278			API.dump(cnf, outpath, otype)
279			except API.UnknownFileTypeError:
280			_exit_with_output(fmsg % outpath, 1)
281			except API.UnknownParserTypeError:
282			_exit_with_output("Invalid output type '%s'" % otype, 1)
283
284
285			def _output_result(cnf, outpath, otype, inpath, itype):
286			"""
287			:param cnf: Configuration object to print out
288			:param outpath: Output file path or None
289			:param otype: Output type or None
290			:param inpath: Input file path
291			:param itype: Input type or None
292			"""
293			fmsg = ("Uknown file type and cannot detect appropriate backend "
294			"from its extension, '%s'")
295
296			if not anyconfig.utils.is_dict_like(cnf):
297			_exit_with_output(str(cnf)) # Print primitive types as it is.
298
299			if not outpath or outpath == "-":
300			outpath = sys.stdout
301			if otype is None:
302			otype = _output_type_by_input_path(inpath, itype, fmsg)
303
304			_try_dump(cnf, outpath, otype, fmsg)
305
306
307			def _load_diff(args):
308			"""
309			:param args: :class:`~argparse.Namespace` object
310			"""
311			try:
312			diff = API.load(args.inputs, args.itype,
313			ignore_missing=args.ignore_missing,
314			ac_merge=args.merge,
315			ac_template=args.template,
316			ac_schema=args.schema)
317			except API.UnknownParserTypeError:
318			_exit_with_output("Wrong input type '%s'" % args.itype, 1)
319			except API.UnknownFileTypeError:
320			_exit_with_output("No appropriate backend was found for given file "
321			"'%s'" % args.itype, 1)
322			_exit_if_load_failure(diff,
323			"Failed to load: args=%s" % ", ".join(args.inputs))
324
325			return diff
326
327
328			def _do_filter(cnf, args):
329			"""
330			:param cnf: Mapping object represents configuration data
331			:param args: :class:`~argparse.Namespace` object
332			:return: `cnf` may be updated
333			"""
334			if args.query:
335			cnf = API.query(cnf, args.query)
336			elif args.get:
337			cnf = _do_get(cnf, args.get)
338			elif args.set:
339			(key, val) = args.set.split('=')
340			API.set_(cnf, key, anyconfig.parser.parse(val))
341
342			return cnf
343
344
345			def main(argv=None):
346			"""
347			:param argv: Argument list to parse or None (sys.argv will be set).
348			"""
349			args = _parse_args((argv if argv else sys.argv)[1:])
350			cnf = os.environ.copy() if args.env else {}
351			diff = _load_diff(args)
352			API.merge(cnf, diff)
353
354			if args.args:
355			diff = anyconfig.parser.parse(args.args)
356			API.merge(cnf, diff)
357
358			if args.validate:
359			_exit_with_output("Validation succeds")
360
361			cnf = API.gen_schema(cnf) if args.gen_schema else _do_filter(cnf, args)
362
363			_output_result(cnf, args.output, args.otype, args.inputs, args.itype)
364
365
366			if __name__ == '__main__':
367			main(sys.argv)
368
369			# vim:sw=4:ts=4:et:
370

ssato / python-anyconfig

Push — master ( dde88c...c42960 )

_do_get() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like