_output_result() - Code Metrics - Inspection of "fix: follow API changes in the frontend and its te..." - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 7f9492...408ecb )

by Satoru

created 2017-02-09 14:22 UTC

_output_result() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	10
c	0
b	0
f	0
dl	0
loc	34
rs	3.1304

How to fix Complexity

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

import codecs
import locale
import logging
import optparse
import os
import sys

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


_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 [Options...] CONF_PATH_OR_PATTERN_0 [CONF_PATH_OR_PATTERN_1 ..]

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

_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'.")
_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 parse_args(argv=None, defaults=None):
    """
    Make up an option and arguments parser.

    :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 = optparse.OptionParser(USAGE, version="%%prog %s" %
                                   anyconfig.globals.VERSION)
    parser.set_defaults(**defaults)

    lpog = optparse.OptionGroup(parser, "List specific options")
    lpog.add_option("-L", "--list", help="List supported config types",
                    action="store_true")
    parser.add_option_group(lpog)

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

    gspog = optparse.OptionGroup(parser, "Get/set options")
    gspog.add_option("", "--get", help=_GET_HELP)
    gspog.add_option("", "--set", help=_SET_HELP)
    parser.add_option_group(gspog)

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

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

    if argv is None:
        argv = sys.argv

    (options, args) = parser.parse_args(argv[1:])
    return (parser, options, args)


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 _check_options_and_args(parser, options, args):
    """
    Show supported config format types or usage.

    :param parser: Option parser object
    :param options: Options optparse.OptionParser.parse_args returns
    :param args: Arguments optparse.OptionParser.parse_args returns
    """
    if not args:
        if options.list:
            tlist = ", ".join(API.list_types())
            _exit_with_output("Supported config types: " + tlist)
        else:
            parser.print_usage()
            sys.exit(1)

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


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 _exit_if_only_to_validate(only_to_validate):
    """
    :param only_to_validate: True if it's only to validate
    """
    if only_to_validate:
        _exit_with_output("Validation succeds")


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_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
    """
    msg = ("Specify inpath and/or outpath type[s] with -I/--itype "
           "or -O/--otype option explicitly")
    fmsg = ("Uknown file type and cannot detect appropriate backend "
            "from its extension, '%s'")
    if not outpath or outpath == "-":
        outpath = sys.stdout
        if otype is None:
            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

    if anyconfig.mdicts.is_dict_like(cnf):
        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)
    else:
        _exit_with_output(str(cnf))  # Output primitive types as it is.


def main(argv=None):
    """
    :param argv: Argument list to parse or None (sys.argv will be set).
    """
    (parser, options, args) = parse_args(argv=argv)
    LOGGER.setLevel(to_log_level(options.loglevel))

    _check_options_and_args(parser, options, args)

    cnf = API.to_container(os.environ.copy() if options.env else {})
    try:
        diff = API.load(args, options.itype,
                        ignore_missing=options.ignore_missing,
                        ac_merge=options.merge,
                        ac_template=options.template,
                        ac_schema=options.schema)
    except API.UnknownParserTypeError:
        _exit_with_output("Wrong input type '%s'" % options.itype, 1)
    except API.UnknownFileTypeError:
        _exit_with_output("No appropriate backend was found for given file "
                          "'%s'" % options.itype, 1)
    _exit_if_load_failure(diff, "Failed to load: args=%s" % ", ".join(args))
    cnf.update(diff)

    if options.args:
        diff = anyconfig.parser.parse(options.args)
        cnf.update(diff)

    _exit_if_only_to_validate(options.validate)

    if options.gen_schema:
        cnf = API.gen_schema(cnf)

    if options.get:
        cnf = _do_get(cnf, options.get)

    if options.set:
        (key, val) = options.set.split('=')
        API.set_(cnf, key, anyconfig.parser.parse(val))

    _output_result(cnf, options.output, options.otype, args[0], options.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 codecs
10			import locale
11			import logging
12			import optparse
13			import os
14			import sys
15
16			import anyconfig.api as API
17			import anyconfig.compat
18			import anyconfig.globals
19			import anyconfig.mdicts
20			import anyconfig.parser
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 [Options...] CONF_PATH_OR_PATTERN_0 [CONF_PATH_OR_PATTERN_1 ..]
47
48			Examples:
49			%prog --list # -> Supported config types: configobj, ini, json, ...
50			# Merge and/or convert input config to output config [file]
51			%prog -I yaml -O yaml /etc/xyz/conf.d/a.conf
52			%prog -I yaml '/etc/xyz/conf.d/*.conf' -o xyz.conf --otype json
53			%prog '/etc/xyz/conf.d/*.json' -o xyz.yml \\
54			--atype json -A '{"obsoletes": "syscnf", "conflicts": "syscnf-old"}'
55			%prog '/etc/xyz/conf.d/*.json' -o xyz.yml \\
56			-A obsoletes:syscnf;conflicts:syscnf-old
57			%prog /etc/foo.json /etc/foo/conf.d/x.json /etc/foo/conf.d/y.json
58			%prog '/etc/foo.d/*.json' -M noreplace
59			# Get/set part of input config
60			%prog '/etc/foo.d/*.json' --get a.b.c
61			%prog '/etc/foo.d/*.json' --set a.b.c=1
62			# Validate with JSON schema or generate JSON schema:
63			%prog --validate -S foo.conf.schema.yml '/etc/foo.d/*.xml'
64			%prog --gen-schema '/etc/foo.d/*.xml' -o foo.conf.schema.yml"""
65
66			DEFAULTS = dict(loglevel=1, list=False, output=None, itype=None,
67			otype=None, atype=None, merge=API.MS_DICTS,
68			ignore_missing=False, template=False, env=False,
69			schema=None, validate=False, gen_schema=False)
70
71
72			def to_log_level(level):
73			"""
74			:param level: Logging level in int = 0 .. 2
75
76			>>> to_log_level(0) == logging.WARN
77			True
78			>>> to_log_level(5) # doctest: +IGNORE_EXCEPTION_DETAIL, +ELLIPSIS
79			Traceback (most recent call last):
80			...
81			ValueError: wrong log level passed: 5
82			>>>
83			"""
84			if not (level >= 0 and level < 3):
85			raise ValueError("wrong log level passed: " + str(level))
86
87			return [logging.WARN, logging.INFO, logging.DEBUG][level]
88
89
90			_ATYPE_HELP_FMT = """\
91			Explicitly select type of argument to provide configs from %s.
92
93			If this option is not set, original parser is used: 'K:V' will become {K: V},
94			'K:V_0,V_1,..' will become {K: [V_0, V_1, ...]}, and 'K_0:V_0;K_1:V_1' will
95			become {K_0: V_0, K_1: V_1} (where the tyep of K is str, type of V is one of
96			Int, str, etc."""
97
98			_GET_HELP = ("Specify key path to get part of config, for example, "
99			"'--get a.b.c' to config {'a': {'b': {'c': 0, 'd': 1}}} "
100			"gives 0 and '--get a.b' to the same config gives "
101			"{'c': 0, 'd': 1}. Path expression can be JSON Pointer "
102			"expression (http://tools.ietf.org/html/rfc6901) such like "
103			"'', '/a~1b', '/m~0n'.")
104			_SET_HELP = ("Specify key path to set (update) part of config, for "
105			"example, '--set a.b.c=1' to a config {'a': {'b': {'c': 0, "
106			"'d': 1}}} gives {'a': {'b': {'c': 1, 'd': 1}}}.")
107
108
109			def parse_args(argv=None, defaults=None):
110			"""
111			Make up an option and arguments parser.
112
113			:param defaults: Default option values
114			"""
115			if defaults is None:
116			defaults = DEFAULTS
117
118			ctypes = API.list_types()
119			ctypes_s = ", ".join(ctypes)
120			type_help = "Select type of %s config files from " + \
121			ctypes_s + " [Automatically detected by file ext]"
122
123			mts = API.MERGE_STRATEGIES
124			mts_s = ", ".join(mts)
125			mt_help = "Select strategy to merge multiple configs from " + \
126			mts_s + " [%(merge)s]" % defaults
127
128			parser = optparse.OptionParser(USAGE, version="%%prog %s" %
129			anyconfig.globals.VERSION)
130			parser.set_defaults(**defaults)
131
132			lpog = optparse.OptionGroup(parser, "List specific options")
133			lpog.add_option("-L", "--list", help="List supported config types",
134			action="store_true")
135			parser.add_option_group(lpog)
136
137			spog = optparse.OptionGroup(parser, "Schema specific options")
138			spog.add_option("", "--validate", action="store_true",
139			help="Only validate input files and do not output. "
140			"You must specify schema file with -S/--schema "
141			"option.")
142			spog.add_option("", "--gen-schema", action="store_true",
143			help="Generate JSON schema for givne config file[s] "
144			"and output it instead of (merged) configuration.")
145			parser.add_option_group(spog)
146
147			gspog = optparse.OptionGroup(parser, "Get/set options")
148			gspog.add_option("", "--get", help=_GET_HELP)
149			gspog.add_option("", "--set", help=_SET_HELP)
150			parser.add_option_group(gspog)
151
152			parser.add_option("-o", "--output", help="Output file path")
153			parser.add_option("-I", "--itype", choices=ctypes,
154			help=(type_help % "Input"))
155			parser.add_option("-O", "--otype", choices=ctypes,
156			help=(type_help % "Output"))
157			parser.add_option("-M", "--merge", choices=mts, help=mt_help)
158			parser.add_option("-A", "--args", help="Argument configs to override")
159			parser.add_option("", "--atype", choices=ctypes,
160			help=_ATYPE_HELP_FMT % ctypes_s)
161
162			parser.add_option("-x", "--ignore-missing", action="store_true",
163			help="Ignore missing input files")
164			parser.add_option("-T", "--template", action="store_true",
165			help="Enable template config support")
166			parser.add_option("-E", "--env", action="store_true",
167			help="Load configuration defaults from "
168			"environment values")
169			parser.add_option("-S", "--schema", help="Specify Schema file[s] path")
170			parser.add_option("-s", "--silent", action="store_const", dest="loglevel",
171			const=0, help="Silent or quiet mode")
172			parser.add_option("-q", "--quiet", action="store_const", dest="loglevel",
173			const=0, help="Same as --silent option")
174			parser.add_option("-v", "--verbose", action="store_const", dest="loglevel",
175			const=2, help="Verbose mode")
176
177			if argv is None:
178			argv = sys.argv
179
180			(options, args) = parser.parse_args(argv[1:])
181			return (parser, options, args)
182
183
184			def _exit_with_output(content, exit_code=0):
185			"""
186			Exit the program with printing out messages.
187
188			:param content: content to print out
189			:param exit_code: Exit code
190			"""
191			(sys.stdout if exit_code == 0 else sys.stderr).write(content + "\n")
192			sys.exit(exit_code)
193
194
195			def _check_options_and_args(parser, options, args):
196			"""
197			Show supported config format types or usage.
198
199			:param parser: Option parser object
200			:param options: Options optparse.OptionParser.parse_args returns
201			:param args: Arguments optparse.OptionParser.parse_args returns
202			"""
203			if not args:
204			if options.list:
205			tlist = ", ".join(API.list_types())
206			_exit_with_output("Supported config types: " + tlist)
207			else:
208			parser.print_usage()
209			sys.exit(1)
210
211			if options.validate and options.schema is None:
212			_exit_with_output("--validate option requires --scheme option", 1)
213
214
215			def _exit_if_load_failure(cnf, msg):
216			"""
217			:param cnf: Loaded configuration object or None indicates load failure
218			:param msg: Message to print out if failure
219			"""
220			if cnf is None:
221			_exit_with_output(msg, 1)
222
223
224			def _exit_if_only_to_validate(only_to_validate):
225			"""
226			:param only_to_validate: True if it's only to validate
227			"""
228			if only_to_validate:
229			_exit_with_output("Validation succeds")
230
231
232			def _do_get(cnf, get_path):
233			"""
234			:param cnf: Configuration object to print out
235			:param get_path: key path given in --get option
236			:return: updated Configuration object if no error
237			"""
238			(cnf, err) = API.get(cnf, get_path)
239			if cnf is None: # Failed to get the result.
240			_exit_with_output("Failed to get result: err=%s" % err, 1)
241
242			return cnf
243
244
245			def _output_result(cnf, outpath, otype, inpath, itype):
246			"""
247			:param cnf: Configuration object to print out
248			:param outpath: Output file path or None
249			:param otype: Output type or None
250			:param inpath: Input file path
251			:param itype: Input type or None
252			"""
253			msg = ("Specify inpath and/or outpath type[s] with -I/--itype "
254			"or -O/--otype option explicitly")
255			fmsg = ("Uknown file type and cannot detect appropriate backend "
256			"from its extension, '%s'")
257			if not outpath or outpath == "-":
258			outpath = sys.stdout
259			if otype is None:
260			if itype is None:
261			try:
262			otype = API.find_loader(inpath).type()
263			except API.UnknownFileTypeError:
264			_exit_with_output((fmsg % inpath) + msg, 1)
265			except ValueError:
266			_exit_with_output(msg, 1)
267			else:
268			otype = itype
269
270			if anyconfig.mdicts.is_dict_like(cnf):
271			try:
272			API.dump(cnf, outpath, otype)
273			except API.UnknownFileTypeError:
274			_exit_with_output(fmsg % outpath, 1)
275			except API.UnknownParserTypeError:
276			_exit_with_output("Invalid output type '%s'" % otype, 1)
277			else:
278			_exit_with_output(str(cnf)) # Output primitive types as it is.
279
280
281			def main(argv=None):
282			"""
283			:param argv: Argument list to parse or None (sys.argv will be set).
284			"""
285			(parser, options, args) = parse_args(argv=argv)
286			LOGGER.setLevel(to_log_level(options.loglevel))
287
288			_check_options_and_args(parser, options, args)
289
290			cnf = API.to_container(os.environ.copy() if options.env else {})
291			try:
292			diff = API.load(args, options.itype,
293			ignore_missing=options.ignore_missing,
294			ac_merge=options.merge,
295			ac_template=options.template,
296			ac_schema=options.schema)
297			except API.UnknownParserTypeError:
298			_exit_with_output("Wrong input type '%s'" % options.itype, 1)
299			except API.UnknownFileTypeError:
300			_exit_with_output("No appropriate backend was found for given file "
301			"'%s'" % options.itype, 1)
302			_exit_if_load_failure(diff, "Failed to load: args=%s" % ", ".join(args))
303			cnf.update(diff)
304
305			if options.args:
306			diff = anyconfig.parser.parse(options.args)
307			cnf.update(diff)
308
309			_exit_if_only_to_validate(options.validate)
310
311			if options.gen_schema:
312			cnf = API.gen_schema(cnf)
313
314			if options.get:
315			cnf = _do_get(cnf, options.get)
316
317			if options.set:
318			(key, val) = options.set.split('=')
319			API.set_(cnf, key, anyconfig.parser.parse(val))
320
321			_output_result(cnf, options.output, options.otype, args[0], options.itype)
322
323
324			if __name__ == '__main__':
325			main(sys.argv)
326
327			# vim:sw=4:ts=4:et:
328

ssato / python-anyconfig

Push — master ( 7f9492...408ecb )

_output_result() F

Complexity

Size

Duplication

Importance

How to fix Complexity

Complexity

Duplication Side-by-Side

Filter issues like