_parse_args() - Code Metrics - ssato/python-anyconfig - Measure and Improve Code Quality continuously with Scrutinizer

_parse_args() B
last analyzed 2018-07-04 11:50 UTC

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	6
dl	0
loc	27
rs	8.2986
c	0
b	0
f	0

#
# 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] or 'UTF-8'

logging.basicConfig(format="%(levelname)s: %(message)s")
LOGGER = logging.getLogger("anyconfig")
LOGGER.addHandler(logging.StreamHandler())

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=0, 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=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, metavar="ITYPE",
                        help=(type_help % "Input"))
    parser.add_argument("-O", "--otype", choices=ctypes, metavar="OTYPE",
                        help=(type_help % "Output"))
    parser.add_argument("-M", "--merge", choices=mts, metavar="MERGE",
                        help=mt_help)
    parser.add_argument("-A", "--args", help="Argument configs to override")
    parser.add_argument("--atype", choices=ctypes, metavar="ATYPE",
                        help=_ATYPE_HELP_FMT % ctypes_s)

    cpog = parser.add_argument_group("Common options")
    cpog.add_argument("-x", "--ignore-missing", action="store_true",
                      help="Ignore missing input files")
    cpog.add_argument("-T", "--template", action="store_true",
                      help="Enable template config support")
    cpog.add_argument("-E", "--env", action="store_true",
                      help="Load configuration defaults from "
                           "environment values")
    cpog.add_argument("-S", "--schema", help="Specify Schema file[s] path")
    cpog.add_argument("-v", "--verbose", action="count", dest="loglevel",
                      help="Verbose mode; -v or -vv (more verbose)")
    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 + os.linesep)
    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(inpaths, itype, fmsg):
    """
    :param inpaths: List of input file paths
    :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(inpaths[0]).type()
        except API.UnknownFileTypeError:
            _exit_with_output((fmsg % inpaths[0]) + msg, 1)
        except (ValueError, IndexError):
            _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.UnknownProcessorTypeError:
        _exit_with_output("Invalid output type '%s'" % otype, 1)


def _output_result(cnf, outpath, otype, inpaths, itype):
    """
    :param cnf: Configuration object to print out
    :param outpath: Output file path or None
    :param otype: Output type or None
    :param inpaths: List of input file paths
    :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(inpaths, 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,
                        ac_ignore_missing=args.ignore_missing,
                        ac_merge=args.merge,
                        ac_template=args.template,
                        ac_schema=args.schema)
    except API.UnknownProcessorTypeError:
        _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)

    if cnf:
        API.merge(cnf, diff)
    else:
        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] or 'UTF-8'
24
25			logging.basicConfig(format="%(levelname)s: %(message)s")
26			LOGGER = logging.getLogger("anyconfig")
27			LOGGER.addHandler(logging.StreamHandler())
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=0, 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=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, metavar="ITYPE",
161			help=(type_help % "Input"))
162			parser.add_argument("-O", "--otype", choices=ctypes, metavar="OTYPE",
163			help=(type_help % "Output"))
164			parser.add_argument("-M", "--merge", choices=mts, metavar="MERGE",
165			help=mt_help)
166			parser.add_argument("-A", "--args", help="Argument configs to override")
167			parser.add_argument("--atype", choices=ctypes, metavar="ATYPE",
168			help=_ATYPE_HELP_FMT % ctypes_s)
169
170			cpog = parser.add_argument_group("Common options")
171			cpog.add_argument("-x", "--ignore-missing", action="store_true",
172			help="Ignore missing input files")
173			cpog.add_argument("-T", "--template", action="store_true",
174			help="Enable template config support")
175			cpog.add_argument("-E", "--env", action="store_true",
176			help="Load configuration defaults from "
177			"environment values")
178			cpog.add_argument("-S", "--schema", help="Specify Schema file[s] path")
179			cpog.add_argument("-v", "--verbose", action="count", dest="loglevel",
180			help="Verbose mode; -v or -vv (more verbose)")
181			return parser
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 + os.linesep)
192			sys.exit(exit_code)
193
194
195			def _parse_args(argv):
196			"""
197			Show supported config format types or usage.
198
199			:param argv: Argument list to parse or None (sys.argv will be set).
200			:return: argparse.Namespace object or None (exit before return)
201			"""
202			parser = make_parser()
203			args = parser.parse_args(argv)
204			LOGGER.setLevel(to_log_level(args.loglevel))
205
206			if not args.inputs:
207			if args.list:
208			tlist = ", ".join(API.list_types())
209			_exit_with_output("Supported config types: " + tlist)
210			elif args.env:
211			cnf = os.environ.copy()
212			_output_result(cnf, args.output, args.otype or "json", None, None)
213			sys.exit(0)
214			else:
215			parser.print_usage()
216			sys.exit(1)
217
218			if args.validate and args.schema is None:
219			_exit_with_output("--validate option requires --scheme option", 1)
220
221			return args
222
223
224			def _exit_if_load_failure(cnf, msg):
225			"""
226			:param cnf: Loaded configuration object or None indicates load failure
227			:param msg: Message to print out if failure
228			"""
229			if cnf is None:
230			_exit_with_output(msg, 1)
231
232
233			def _do_get(cnf, get_path):
234			"""
235			:param cnf: Configuration object to print out
236			:param get_path: key path given in --get option
237			:return: updated Configuration object if no error
238			"""
239			(cnf, err) = API.get(cnf, get_path)
240			if cnf is None: # Failed to get the result.
241			_exit_with_output("Failed to get result: err=%s" % err, 1)
242
243			return cnf
244
245
246			def _output_type_by_input_path(inpaths, itype, fmsg):
247			"""
248			:param inpaths: List of input file paths
249			:param itype: Input type or None
250			:param fmsg: message if it cannot detect otype by `inpath`
251			:return: Output type :: str
252			"""
253			msg = ("Specify inpath and/or outpath type[s] with -I/--itype "
254			"or -O/--otype option explicitly")
255			if itype is None:
256			try:
257			otype = API.find_loader(inpaths[0]).type()
258			except API.UnknownFileTypeError:
259			_exit_with_output((fmsg % inpaths[0]) + msg, 1)
260			except (ValueError, IndexError):
261			_exit_with_output(msg, 1)
262			else:
263			otype = itype
264
265			return otype
266
267
268			def _try_dump(cnf, outpath, otype, fmsg):
269			"""
270			:param cnf: Configuration object to print out
271			:param outpath: Output file path or None
272			:param otype: Output type or None
273			:param fmsg: message if it cannot detect otype by `inpath`
274			"""
275			try:
276			API.dump(cnf, outpath, otype)
277			except API.UnknownFileTypeError:
278			_exit_with_output(fmsg % outpath, 1)
279			except API.UnknownProcessorTypeError:
280			_exit_with_output("Invalid output type '%s'" % otype, 1)
281
282
283			def _output_result(cnf, outpath, otype, inpaths, itype):
284			"""
285			:param cnf: Configuration object to print out
286			:param outpath: Output file path or None
287			:param otype: Output type or None
288			:param inpaths: List of input file paths
289			:param itype: Input type or None
290			"""
291			fmsg = ("Uknown file type and cannot detect appropriate backend "
292			"from its extension, '%s'")
293
294			if not anyconfig.utils.is_dict_like(cnf):
295			_exit_with_output(str(cnf)) # Print primitive types as it is.
296
297			if not outpath or outpath == "-":
298			outpath = sys.stdout
299			if otype is None:
300			otype = _output_type_by_input_path(inpaths, itype, fmsg)
301
302			_try_dump(cnf, outpath, otype, fmsg)
303
304
305			def _load_diff(args):
306			"""
307			:param args: :class:`~argparse.Namespace` object
308			"""
309			try:
310			diff = API.load(args.inputs, args.itype,
311			ac_ignore_missing=args.ignore_missing,
312			ac_merge=args.merge,
313			ac_template=args.template,
314			ac_schema=args.schema)
315			except API.UnknownProcessorTypeError:
316			_exit_with_output("Wrong input type '%s'" % args.itype, 1)
317			except API.UnknownFileTypeError:
318			_exit_with_output("No appropriate backend was found for given file "
319			"'%s'" % args.itype, 1)
320			_exit_if_load_failure(diff,
321			"Failed to load: args=%s" % ", ".join(args.inputs))
322
323			return diff
324
325
326			def _do_filter(cnf, args):
327			"""
328			:param cnf: Mapping object represents configuration data
329			:param args: :class:`~argparse.Namespace` object
330			:return: `cnf` may be updated
331			"""
332			if args.query:
333			cnf = API.query(cnf, args.query)
334			elif args.get:
335			cnf = _do_get(cnf, args.get)
336			elif args.set:
337			(key, val) = args.set.split('=')
338			API.set_(cnf, key, anyconfig.parser.parse(val))
339
340			return cnf
341
342
343			def main(argv=None):
344			"""
345			:param argv: Argument list to parse or None (sys.argv will be set).
346			"""
347			args = _parse_args((argv if argv else sys.argv)[1:])
348			cnf = os.environ.copy() if args.env else {}
349			diff = _load_diff(args)
350
351			if cnf:
352			API.merge(cnf, diff)
353			else:
354			cnf = diff
355
356			if args.args:
357			diff = anyconfig.parser.parse(args.args)
358			API.merge(cnf, diff)
359
360			if args.validate:
361			_exit_with_output("Validation succeds")
362
363			cnf = API.gen_schema(cnf) if args.gen_schema else _do_filter(cnf, args)
364			_output_result(cnf, args.output, args.otype, args.inputs, args.itype)
365
366
367			if __name__ == '__main__':
368			main(sys.argv)
369
370			# vim:sw=4:ts=4:et:
371

ssato / python-anyconfig

_parse_args() B last analyzed 2018-07-04 11:50 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

_parse_args() B
last analyzed 2018-07-04 11:50 UTC