diff options
Diffstat (limited to 'chromium/third_party/logilab/logilab/common/configuration.py')
| -rw-r--r-- | chromium/third_party/logilab/logilab/common/configuration.py | 1105 |
1 files changed, 0 insertions, 1105 deletions
diff --git a/chromium/third_party/logilab/logilab/common/configuration.py b/chromium/third_party/logilab/logilab/common/configuration.py deleted file mode 100644 index b2924277c94..00000000000 --- a/chromium/third_party/logilab/logilab/common/configuration.py +++ /dev/null @@ -1,1105 +0,0 @@ -# copyright 2003-2012 LOGILAB S.A. (Paris, FRANCE), all rights reserved. -# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr -# -# This file is part of logilab-common. -# -# logilab-common is free software: you can redistribute it and/or modify it under -# the terms of the GNU Lesser General Public License as published by the Free -# Software Foundation, either version 2.1 of the License, or (at your option) any -# later version. -# -# logilab-common is distributed in the hope that it will be useful, but WITHOUT -# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS -# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more -# details. -# -# You should have received a copy of the GNU Lesser General Public License along -# with logilab-common. If not, see <http://www.gnu.org/licenses/>. -"""Classes to handle advanced configuration in simple to complex applications. - -Allows to load the configuration from a file or from command line -options, to generate a sample configuration file or to display -program's usage. Fills the gap between optik/optparse and ConfigParser -by adding data types (which are also available as a standalone optik -extension in the `optik_ext` module). - - -Quick start: simplest usage ---------------------------- - -.. python :: - - >>> import sys - >>> from logilab.common.configuration import Configuration - >>> options = [('dothis', {'type':'yn', 'default': True, 'metavar': '<y or n>'}), - ... ('value', {'type': 'string', 'metavar': '<string>'}), - ... ('multiple', {'type': 'csv', 'default': ('yop',), - ... 'metavar': '<comma separated values>', - ... 'help': 'you can also document the option'}), - ... ('number', {'type': 'int', 'default':2, 'metavar':'<int>'}), - ... ] - >>> config = Configuration(options=options, name='My config') - >>> print config['dothis'] - True - >>> print config['value'] - None - >>> print config['multiple'] - ('yop',) - >>> print config['number'] - 2 - >>> print config.help() - Usage: [options] - - Options: - -h, --help show this help message and exit - --dothis=<y or n> - --value=<string> - --multiple=<comma separated values> - you can also document the option [current: none] - --number=<int> - - >>> f = open('myconfig.ini', 'w') - >>> f.write('''[MY CONFIG] - ... number = 3 - ... dothis = no - ... multiple = 1,2,3 - ... ''') - >>> f.close() - >>> config.load_file_configuration('myconfig.ini') - >>> print config['dothis'] - False - >>> print config['value'] - None - >>> print config['multiple'] - ['1', '2', '3'] - >>> print config['number'] - 3 - >>> sys.argv = ['mon prog', '--value', 'bacon', '--multiple', '4,5,6', - ... 'nonoptionargument'] - >>> print config.load_command_line_configuration() - ['nonoptionargument'] - >>> print config['value'] - bacon - >>> config.generate_config() - # class for simple configurations which don't need the - # manager / providers model and prefer delegation to inheritance - # - # configuration values are accessible through a dict like interface - # - [MY CONFIG] - - dothis=no - - value=bacon - - # you can also document the option - multiple=4,5,6 - - number=3 - - Note : starting with Python 2.7 ConfigParser is able to take into - account the order of occurrences of the options into a file (by - using an OrderedDict). If you have two options changing some common - state, like a 'disable-all-stuff' and a 'enable-some-stuff-a', their - order of appearance will be significant : the last specified in the - file wins. For earlier version of python and logilab.common newer - than 0.61 the behaviour is unspecified. - -""" - -from __future__ import print_function - -__docformat__ = "restructuredtext en" - -__all__ = ('OptionsManagerMixIn', 'OptionsProviderMixIn', - 'ConfigurationMixIn', 'Configuration', - 'OptionsManager2ConfigurationAdapter') - -import os -import sys -import re -from os.path import exists, expanduser -from copy import copy -from warnings import warn - -from six import string_types -from six.moves import range, configparser as cp, input - -from logilab.common.compat import str_encode as _encode -from logilab.common.deprecation import deprecated -from logilab.common.textutils import normalize_text, unquote -from logilab.common import optik_ext - -OptionError = optik_ext.OptionError - -REQUIRED = [] - -class UnsupportedAction(Exception): - """raised by set_option when it doesn't know what to do for an action""" - - -def _get_encoding(encoding, stream): - encoding = encoding or getattr(stream, 'encoding', None) - if not encoding: - import locale - encoding = locale.getpreferredencoding() - return encoding - - -# validation functions ######################################################## - -# validators will return the validated value or raise optparse.OptionValueError -# XXX add to documentation - -def choice_validator(optdict, name, value): - """validate and return a converted value for option of type 'choice' - """ - if not value in optdict['choices']: - msg = "option %s: invalid value: %r, should be in %s" - raise optik_ext.OptionValueError(msg % (name, value, optdict['choices'])) - return value - -def multiple_choice_validator(optdict, name, value): - """validate and return a converted value for option of type 'choice' - """ - choices = optdict['choices'] - values = optik_ext.check_csv(None, name, value) - for value in values: - if not value in choices: - msg = "option %s: invalid value: %r, should be in %s" - raise optik_ext.OptionValueError(msg % (name, value, choices)) - return values - -def csv_validator(optdict, name, value): - """validate and return a converted value for option of type 'csv' - """ - return optik_ext.check_csv(None, name, value) - -def yn_validator(optdict, name, value): - """validate and return a converted value for option of type 'yn' - """ - return optik_ext.check_yn(None, name, value) - -def named_validator(optdict, name, value): - """validate and return a converted value for option of type 'named' - """ - return optik_ext.check_named(None, name, value) - -def file_validator(optdict, name, value): - """validate and return a filepath for option of type 'file'""" - return optik_ext.check_file(None, name, value) - -def color_validator(optdict, name, value): - """validate and return a valid color for option of type 'color'""" - return optik_ext.check_color(None, name, value) - -def password_validator(optdict, name, value): - """validate and return a string for option of type 'password'""" - return optik_ext.check_password(None, name, value) - -def date_validator(optdict, name, value): - """validate and return a mx DateTime object for option of type 'date'""" - return optik_ext.check_date(None, name, value) - -def time_validator(optdict, name, value): - """validate and return a time object for option of type 'time'""" - return optik_ext.check_time(None, name, value) - -def bytes_validator(optdict, name, value): - """validate and return an integer for option of type 'bytes'""" - return optik_ext.check_bytes(None, name, value) - - -VALIDATORS = {'string': unquote, - 'int': int, - 'float': float, - 'file': file_validator, - 'font': unquote, - 'color': color_validator, - 'regexp': re.compile, - 'csv': csv_validator, - 'yn': yn_validator, - 'bool': yn_validator, - 'named': named_validator, - 'password': password_validator, - 'date': date_validator, - 'time': time_validator, - 'bytes': bytes_validator, - 'choice': choice_validator, - 'multiple_choice': multiple_choice_validator, - } - -def _call_validator(opttype, optdict, option, value): - if opttype not in VALIDATORS: - raise Exception('Unsupported type "%s"' % opttype) - try: - return VALIDATORS[opttype](optdict, option, value) - except TypeError: - try: - return VALIDATORS[opttype](value) - except optik_ext.OptionValueError: - raise - except: - raise optik_ext.OptionValueError('%s value (%r) should be of type %s' % - (option, value, opttype)) - -# user input functions ######################################################## - -# user input functions will ask the user for input on stdin then validate -# the result and return the validated value or raise optparse.OptionValueError -# XXX add to documentation - -def input_password(optdict, question='password:'): - from getpass import getpass - while True: - value = getpass(question) - value2 = getpass('confirm: ') - if value == value2: - return value - print('password mismatch, try again') - -def input_string(optdict, question): - value = input(question).strip() - return value or None - -def _make_input_function(opttype): - def input_validator(optdict, question): - while True: - value = input(question) - if not value.strip(): - return None - try: - return _call_validator(opttype, optdict, None, value) - except optik_ext.OptionValueError as ex: - msg = str(ex).split(':', 1)[-1].strip() - print('bad value: %s' % msg) - return input_validator - -INPUT_FUNCTIONS = { - 'string': input_string, - 'password': input_password, - } - -for opttype in VALIDATORS.keys(): - INPUT_FUNCTIONS.setdefault(opttype, _make_input_function(opttype)) - -# utility functions ############################################################ - -def expand_default(self, option): - """monkey patch OptionParser.expand_default since we have a particular - way to handle defaults to avoid overriding values in the configuration - file - """ - if self.parser is None or not self.default_tag: - return option.help - optname = option._long_opts[0][2:] - try: - provider = self.parser.options_manager._all_options[optname] - except KeyError: - value = None - else: - optdict = provider.get_option_def(optname) - optname = provider.option_attrname(optname, optdict) - value = getattr(provider.config, optname, optdict) - value = format_option_value(optdict, value) - if value is optik_ext.NO_DEFAULT or not value: - value = self.NO_DEFAULT_VALUE - return option.help.replace(self.default_tag, str(value)) - - -def _validate(value, optdict, name=''): - """return a validated value for an option according to its type - - optional argument name is only used for error message formatting - """ - try: - _type = optdict['type'] - except KeyError: - # FIXME - return value - return _call_validator(_type, optdict, name, value) -convert = deprecated('[0.60] convert() was renamed _validate()')(_validate) - -# format and output functions ################################################## - -def comment(string): - """return string as a comment""" - lines = [line.strip() for line in string.splitlines()] - return '# ' + ('%s# ' % os.linesep).join(lines) - -def format_time(value): - if not value: - return '0' - if value != int(value): - return '%.2fs' % value - value = int(value) - nbmin, nbsec = divmod(value, 60) - if nbsec: - return '%ss' % value - nbhour, nbmin_ = divmod(nbmin, 60) - if nbmin_: - return '%smin' % nbmin - nbday, nbhour_ = divmod(nbhour, 24) - if nbhour_: - return '%sh' % nbhour - return '%sd' % nbday - -def format_bytes(value): - if not value: - return '0' - if value != int(value): - return '%.2fB' % value - value = int(value) - prevunit = 'B' - for unit in ('KB', 'MB', 'GB', 'TB'): - next, remain = divmod(value, 1024) - if remain: - return '%s%s' % (value, prevunit) - prevunit = unit - value = next - return '%s%s' % (value, unit) - -def format_option_value(optdict, value): - """return the user input's value from a 'compiled' value""" - if isinstance(value, (list, tuple)): - value = ','.join(value) - elif isinstance(value, dict): - value = ','.join(['%s:%s' % (k, v) for k, v in value.items()]) - elif hasattr(value, 'match'): # optdict.get('type') == 'regexp' - # compiled regexp - value = value.pattern - elif optdict.get('type') == 'yn': - value = value and 'yes' or 'no' - elif isinstance(value, string_types) and value.isspace(): - value = "'%s'" % value - elif optdict.get('type') == 'time' and isinstance(value, (float, int, long)): - value = format_time(value) - elif optdict.get('type') == 'bytes' and hasattr(value, '__int__'): - value = format_bytes(value) - return value - -def ini_format_section(stream, section, options, encoding=None, doc=None): - """format an options section using the INI format""" - encoding = _get_encoding(encoding, stream) - if doc: - print(_encode(comment(doc), encoding), file=stream) - print('[%s]' % section, file=stream) - ini_format(stream, options, encoding) - -def ini_format(stream, options, encoding): - """format options using the INI format""" - for optname, optdict, value in options: - value = format_option_value(optdict, value) - help = optdict.get('help') - if help: - help = normalize_text(help, line_len=79, indent='# ') - print(file=stream) - print(_encode(help, encoding), file=stream) - else: - print(file=stream) - if value is None: - print('#%s=' % optname, file=stream) - else: - value = _encode(value, encoding).strip() - print('%s=%s' % (optname, value), file=stream) - -format_section = ini_format_section - -def rest_format_section(stream, section, options, encoding=None, doc=None): - """format an options section using as ReST formatted output""" - encoding = _get_encoding(encoding, stream) - if section: - print('%s\n%s' % (section, "'"*len(section)), file=stream) - if doc: - print(_encode(normalize_text(doc, line_len=79, indent=''), encoding), file=stream) - print(file=stream) - for optname, optdict, value in options: - help = optdict.get('help') - print(':%s:' % optname, file=stream) - if help: - help = normalize_text(help, line_len=79, indent=' ') - print(_encode(help, encoding), file=stream) - if value: - value = _encode(format_option_value(optdict, value), encoding) - print(file=stream) - print(' Default: ``%s``' % value.replace("`` ", "```` ``"), file=stream) - -# Options Manager ############################################################## - -class OptionsManagerMixIn(object): - """MixIn to handle a configuration from both a configuration file and - command line options - """ - - def __init__(self, usage, config_file=None, version=None, quiet=0): - self.config_file = config_file - self.reset_parsers(usage, version=version) - # list of registered options providers - self.options_providers = [] - # dictionary associating option name to checker - self._all_options = {} - self._short_options = {} - self._nocallback_options = {} - self._mygroups = dict() - # verbosity - self.quiet = quiet - self._maxlevel = 0 - - def reset_parsers(self, usage='', version=None): - # configuration file parser - self.cfgfile_parser = cp.ConfigParser() - # command line parser - self.cmdline_parser = optik_ext.OptionParser(usage=usage, version=version) - self.cmdline_parser.options_manager = self - self._optik_option_attrs = set(self.cmdline_parser.option_class.ATTRS) - - def register_options_provider(self, provider, own_group=True): - """register an options provider""" - assert provider.priority <= 0, "provider's priority can't be >= 0" - for i in range(len(self.options_providers)): - if provider.priority > self.options_providers[i].priority: - self.options_providers.insert(i, provider) - break - else: - self.options_providers.append(provider) - non_group_spec_options = [option for option in provider.options - if 'group' not in option[1]] - groups = getattr(provider, 'option_groups', ()) - if own_group and non_group_spec_options: - self.add_option_group(provider.name.upper(), provider.__doc__, - non_group_spec_options, provider) - else: - for opt, optdict in non_group_spec_options: - self.add_optik_option(provider, self.cmdline_parser, opt, optdict) - for gname, gdoc in groups: - gname = gname.upper() - goptions = [option for option in provider.options - if option[1].get('group', '').upper() == gname] - self.add_option_group(gname, gdoc, goptions, provider) - - def add_option_group(self, group_name, doc, options, provider): - """add an option group including the listed options - """ - assert options - # add option group to the command line parser - if group_name in self._mygroups: - group = self._mygroups[group_name] - else: - group = optik_ext.OptionGroup(self.cmdline_parser, - title=group_name.capitalize()) - self.cmdline_parser.add_option_group(group) - group.level = provider.level - self._mygroups[group_name] = group - # add section to the config file - if group_name != "DEFAULT": - self.cfgfile_parser.add_section(group_name) - # add provider's specific options - for opt, optdict in options: - self.add_optik_option(provider, group, opt, optdict) - - def add_optik_option(self, provider, optikcontainer, opt, optdict): - if 'inputlevel' in optdict: - warn('[0.50] "inputlevel" in option dictionary for %s is deprecated,' - ' use "level"' % opt, DeprecationWarning) - optdict['level'] = optdict.pop('inputlevel') - args, optdict = self.optik_option(provider, opt, optdict) - option = optikcontainer.add_option(*args, **optdict) - self._all_options[opt] = provider - self._maxlevel = max(self._maxlevel, option.level or 0) - - def optik_option(self, provider, opt, optdict): - """get our personal option definition and return a suitable form for - use with optik/optparse - """ - optdict = copy(optdict) - others = {} - if 'action' in optdict: - self._nocallback_options[provider] = opt - else: - optdict['action'] = 'callback' - optdict['callback'] = self.cb_set_provider_option - # default is handled here and *must not* be given to optik if you - # want the whole machinery to work - if 'default' in optdict: - if ('help' in optdict - and optdict.get('default') is not None - and not optdict['action'] in ('store_true', 'store_false')): - optdict['help'] += ' [current: %default]' - del optdict['default'] - args = ['--' + str(opt)] - if 'short' in optdict: - self._short_options[optdict['short']] = opt - args.append('-' + optdict['short']) - del optdict['short'] - # cleanup option definition dict before giving it to optik - for key in list(optdict.keys()): - if not key in self._optik_option_attrs: - optdict.pop(key) - return args, optdict - - def cb_set_provider_option(self, option, opt, value, parser): - """optik callback for option setting""" - if opt.startswith('--'): - # remove -- on long option - opt = opt[2:] - else: - # short option, get its long equivalent - opt = self._short_options[opt[1:]] - # trick since we can't set action='store_true' on options - if value is None: - value = 1 - self.global_set_option(opt, value) - - def global_set_option(self, opt, value): - """set option on the correct option provider""" - self._all_options[opt].set_option(opt, value) - - def generate_config(self, stream=None, skipsections=(), encoding=None): - """write a configuration file according to the current configuration - into the given stream or stdout - """ - options_by_section = {} - sections = [] - for provider in self.options_providers: - for section, options in provider.options_by_section(): - if section is None: - section = provider.name - if section in skipsections: - continue - options = [(n, d, v) for (n, d, v) in options - if d.get('type') is not None] - if not options: - continue - if not section in sections: - sections.append(section) - alloptions = options_by_section.setdefault(section, []) - alloptions += options - stream = stream or sys.stdout - encoding = _get_encoding(encoding, stream) - printed = False - for section in sections: - if printed: - print('\n', file=stream) - format_section(stream, section.upper(), options_by_section[section], - encoding) - printed = True - - def generate_manpage(self, pkginfo, section=1, stream=None): - """write a man page for the current configuration into the given - stream or stdout - """ - self._monkeypatch_expand_default() - try: - optik_ext.generate_manpage(self.cmdline_parser, pkginfo, - section, stream=stream or sys.stdout, - level=self._maxlevel) - finally: - self._unmonkeypatch_expand_default() - - # initialization methods ################################################## - - def load_provider_defaults(self): - """initialize configuration using default values""" - for provider in self.options_providers: - provider.load_defaults() - - def load_file_configuration(self, config_file=None): - """load the configuration from file""" - self.read_config_file(config_file) - self.load_config_file() - - def read_config_file(self, config_file=None): - """read the configuration file but do not load it (i.e. dispatching - values to each options provider) - """ - helplevel = 1 - while helplevel <= self._maxlevel: - opt = '-'.join(['long'] * helplevel) + '-help' - if opt in self._all_options: - break # already processed - def helpfunc(option, opt, val, p, level=helplevel): - print(self.help(level)) - sys.exit(0) - helpmsg = '%s verbose help.' % ' '.join(['more'] * helplevel) - optdict = {'action' : 'callback', 'callback' : helpfunc, - 'help' : helpmsg} - provider = self.options_providers[0] - self.add_optik_option(provider, self.cmdline_parser, opt, optdict) - provider.options += ( (opt, optdict), ) - helplevel += 1 - if config_file is None: - config_file = self.config_file - if config_file is not None: - config_file = expanduser(config_file) - if config_file and exists(config_file): - parser = self.cfgfile_parser - parser.read([config_file]) - # normalize sections'title - for sect, values in parser._sections.items(): - if not sect.isupper() and values: - parser._sections[sect.upper()] = values - elif not self.quiet: - msg = 'No config file found, using default configuration' - print(msg, file=sys.stderr) - return - - def input_config(self, onlysection=None, inputlevel=0, stream=None): - """interactively get configuration values by asking to the user and generate - a configuration file - """ - if onlysection is not None: - onlysection = onlysection.upper() - for provider in self.options_providers: - for section, option, optdict in provider.all_options(): - if onlysection is not None and section != onlysection: - continue - if not 'type' in optdict: - # ignore action without type (callback, store_true...) - continue - provider.input_option(option, optdict, inputlevel) - # now we can generate the configuration file - if stream is not None: - self.generate_config(stream) - - def load_config_file(self): - """dispatch values previously read from a configuration file to each - options provider) - """ - parser = self.cfgfile_parser - for section in parser.sections(): - for option, value in parser.items(section): - try: - self.global_set_option(option, value) - except (KeyError, OptionError): - # TODO handle here undeclared options appearing in the config file - continue - - def load_configuration(self, **kwargs): - """override configuration according to given parameters - """ - for opt, opt_value in kwargs.items(): - opt = opt.replace('_', '-') - provider = self._all_options[opt] - provider.set_option(opt, opt_value) - - def load_command_line_configuration(self, args=None): - """override configuration according to command line parameters - - return additional arguments - """ - self._monkeypatch_expand_default() - try: - if args is None: - args = sys.argv[1:] - else: - args = list(args) - (options, args) = self.cmdline_parser.parse_args(args=args) - for provider in self._nocallback_options.keys(): - config = provider.config - for attr in config.__dict__.keys(): - value = getattr(options, attr, None) - if value is None: - continue - setattr(config, attr, value) - return args - finally: - self._unmonkeypatch_expand_default() - - - # help methods ############################################################ - - def add_help_section(self, title, description, level=0): - """add a dummy option section for help purpose """ - group = optik_ext.OptionGroup(self.cmdline_parser, - title=title.capitalize(), - description=description) - group.level = level - self._maxlevel = max(self._maxlevel, level) - self.cmdline_parser.add_option_group(group) - - def _monkeypatch_expand_default(self): - # monkey patch optik_ext to deal with our default values - try: - self.__expand_default_backup = optik_ext.HelpFormatter.expand_default - optik_ext.HelpFormatter.expand_default = expand_default - except AttributeError: - # python < 2.4: nothing to be done - pass - def _unmonkeypatch_expand_default(self): - # remove monkey patch - if hasattr(optik_ext.HelpFormatter, 'expand_default'): - # unpatch optik_ext to avoid side effects - optik_ext.HelpFormatter.expand_default = self.__expand_default_backup - - def help(self, level=0): - """return the usage string for available options """ - self.cmdline_parser.formatter.output_level = level - self._monkeypatch_expand_default() - try: - return self.cmdline_parser.format_help() - finally: - self._unmonkeypatch_expand_default() - - -class Method(object): - """used to ease late binding of default method (so you can define options - on the class using default methods on the configuration instance) - """ - def __init__(self, methname): - self.method = methname - self._inst = None - - def bind(self, instance): - """bind the method to its instance""" - if self._inst is None: - self._inst = instance - - def __call__(self, *args, **kwargs): - assert self._inst, 'unbound method' - return getattr(self._inst, self.method)(*args, **kwargs) - -# Options Provider ############################################################# - -class OptionsProviderMixIn(object): - """Mixin to provide options to an OptionsManager""" - - # those attributes should be overridden - priority = -1 - name = 'default' - options = () - level = 0 - - def __init__(self): - self.config = optik_ext.Values() - for option in self.options: - try: - option, optdict = option - except ValueError: - raise Exception('Bad option: %r' % option) - if isinstance(optdict.get('default'), Method): - optdict['default'].bind(self) - elif isinstance(optdict.get('callback'), Method): - optdict['callback'].bind(self) - self.load_defaults() - - def load_defaults(self): - """initialize the provider using default values""" - for opt, optdict in self.options: - action = optdict.get('action') - if action != 'callback': - # callback action have no default - default = self.option_default(opt, optdict) - if default is REQUIRED: - continue - self.set_option(opt, default, action, optdict) - - def option_default(self, opt, optdict=None): - """return the default value for an option""" - if optdict is None: - optdict = self.get_option_def(opt) - default = optdict.get('default') - if callable(default): - default = default() - return default - - def option_attrname(self, opt, optdict=None): - """get the config attribute corresponding to opt - """ - if optdict is None: - optdict = self.get_option_def(opt) - return optdict.get('dest', opt.replace('-', '_')) - option_name = deprecated('[0.60] OptionsProviderMixIn.option_name() was renamed to option_attrname()')(option_attrname) - - def option_value(self, opt): - """get the current value for the given option""" - return getattr(self.config, self.option_attrname(opt), None) - - def set_option(self, opt, value, action=None, optdict=None): - """method called to set an option (registered in the options list) - """ - if optdict is None: - optdict = self.get_option_def(opt) - if value is not None: - value = _validate(value, optdict, opt) - if action is None: - action = optdict.get('action', 'store') - if optdict.get('type') == 'named': # XXX need specific handling - optname = self.option_attrname(opt, optdict) - currentvalue = getattr(self.config, optname, None) - if currentvalue: - currentvalue.update(value) - value = currentvalue - if action == 'store': - setattr(self.config, self.option_attrname(opt, optdict), value) - elif action in ('store_true', 'count'): - setattr(self.config, self.option_attrname(opt, optdict), 0) - elif action == 'store_false': - setattr(self.config, self.option_attrname(opt, optdict), 1) - elif action == 'append': - opt = self.option_attrname(opt, optdict) - _list = getattr(self.config, opt, None) - if _list is None: - if isinstance(value, (list, tuple)): - _list = value - elif value is not None: - _list = [] - _list.append(value) - setattr(self.config, opt, _list) - elif isinstance(_list, tuple): - setattr(self.config, opt, _list + (value,)) - else: - _list.append(value) - elif action == 'callback': - optdict['callback'](None, opt, value, None) - else: - raise UnsupportedAction(action) - - def input_option(self, option, optdict, inputlevel=99): - default = self.option_default(option, optdict) - if default is REQUIRED: - defaultstr = '(required): ' - elif optdict.get('level', 0) > inputlevel: - return - elif optdict['type'] == 'password' or default is None: - defaultstr = ': ' - else: - defaultstr = '(default: %s): ' % format_option_value(optdict, default) - print(':%s:' % option) - print(optdict.get('help') or option) - inputfunc = INPUT_FUNCTIONS[optdict['type']] - value = inputfunc(optdict, defaultstr) - while default is REQUIRED and not value: - print('please specify a value') - value = inputfunc(optdict, '%s: ' % option) - if value is None and default is not None: - value = default - self.set_option(option, value, optdict=optdict) - - def get_option_def(self, opt): - """return the dictionary defining an option given it's name""" - assert self.options - for option in self.options: - if option[0] == opt: - return option[1] - raise OptionError('no such option %s in section %r' - % (opt, self.name), opt) - - - def all_options(self): - """return an iterator on available options for this provider - option are actually described by a 3-uple: - (section, option name, option dictionary) - """ - for section, options in self.options_by_section(): - if section is None: - if self.name is None: - continue - section = self.name.upper() - for option, optiondict, value in options: - yield section, option, optiondict - - def options_by_section(self): - """return an iterator on options grouped by section - - (section, [list of (optname, optdict, optvalue)]) - """ - sections = {} - for optname, optdict in self.options: - sections.setdefault(optdict.get('group'), []).append( - (optname, optdict, self.option_value(optname))) - if None in sections: - yield None, sections.pop(None) - for section, options in sections.items(): - yield section.upper(), options - - def options_and_values(self, options=None): - if options is None: - options = self.options - for optname, optdict in options: - yield (optname, optdict, self.option_value(optname)) - -# configuration ################################################################ - -class ConfigurationMixIn(OptionsManagerMixIn, OptionsProviderMixIn): - """basic mixin for simple configurations which don't need the - manager / providers model - """ - def __init__(self, *args, **kwargs): - if not args: - kwargs.setdefault('usage', '') - kwargs.setdefault('quiet', 1) - OptionsManagerMixIn.__init__(self, *args, **kwargs) - OptionsProviderMixIn.__init__(self) - if not getattr(self, 'option_groups', None): - self.option_groups = [] - for option, optdict in self.options: - try: - gdef = (optdict['group'].upper(), '') - except KeyError: - continue - if not gdef in self.option_groups: - self.option_groups.append(gdef) - self.register_options_provider(self, own_group=False) - - def register_options(self, options): - """add some options to the configuration""" - options_by_group = {} - for optname, optdict in options: - options_by_group.setdefault(optdict.get('group', self.name.upper()), []).append((optname, optdict)) - for group, options in options_by_group.items(): - self.add_option_group(group, None, options, self) - self.options += tuple(options) - - def load_defaults(self): - OptionsProviderMixIn.load_defaults(self) - - def __iter__(self): - return iter(self.config.__dict__.iteritems()) - - def __getitem__(self, key): - try: - return getattr(self.config, self.option_attrname(key)) - except (optik_ext.OptionValueError, AttributeError): - raise KeyError(key) - - def __setitem__(self, key, value): - self.set_option(key, value) - - def get(self, key, default=None): - try: - return getattr(self.config, self.option_attrname(key)) - except (OptionError, AttributeError): - return default - - -class Configuration(ConfigurationMixIn): - """class for simple configurations which don't need the - manager / providers model and prefer delegation to inheritance - - configuration values are accessible through a dict like interface - """ - - def __init__(self, config_file=None, options=None, name=None, - usage=None, doc=None, version=None): - if options is not None: - self.options = options - if name is not None: - self.name = name - if doc is not None: - self.__doc__ = doc - super(Configuration, self).__init__(config_file=config_file, usage=usage, version=version) - - -class OptionsManager2ConfigurationAdapter(object): - """Adapt an option manager to behave like a - `logilab.common.configuration.Configuration` instance - """ - def __init__(self, provider): - self.config = provider - - def __getattr__(self, key): - return getattr(self.config, key) - - def __getitem__(self, key): - provider = self.config._all_options[key] - try: - return getattr(provider.config, provider.option_attrname(key)) - except AttributeError: - raise KeyError(key) - - def __setitem__(self, key, value): - self.config.global_set_option(self.config.option_attrname(key), value) - - def get(self, key, default=None): - provider = self.config._all_options[key] - try: - return getattr(provider.config, provider.option_attrname(key)) - except AttributeError: - return default - -# other functions ############################################################## - -def read_old_config(newconfig, changes, configfile): - """initialize newconfig from a deprecated configuration file - - possible changes: - * ('renamed', oldname, newname) - * ('moved', option, oldgroup, newgroup) - * ('typechanged', option, oldtype, newvalue) - """ - # build an index of changes - changesindex = {} - for action in changes: - if action[0] == 'moved': - option, oldgroup, newgroup = action[1:] - changesindex.setdefault(option, []).append((action[0], oldgroup, newgroup)) - continue - if action[0] == 'renamed': - oldname, newname = action[1:] - changesindex.setdefault(newname, []).append((action[0], oldname)) - continue - if action[0] == 'typechanged': - option, oldtype, newvalue = action[1:] - changesindex.setdefault(option, []).append((action[0], oldtype, newvalue)) - continue - if action[1] in ('added', 'removed'): - continue # nothing to do here - raise Exception('unknown change %s' % action[0]) - # build a config object able to read the old config - options = [] - for optname, optdef in newconfig.options: - for action in changesindex.pop(optname, ()): - if action[0] == 'moved': - oldgroup, newgroup = action[1:] - optdef = optdef.copy() - optdef['group'] = oldgroup - elif action[0] == 'renamed': - optname = action[1] - elif action[0] == 'typechanged': - oldtype = action[1] - optdef = optdef.copy() - optdef['type'] = oldtype - options.append((optname, optdef)) - if changesindex: - raise Exception('unapplied changes: %s' % changesindex) - oldconfig = Configuration(options=options, name=newconfig.name) - # read the old config - oldconfig.load_file_configuration(configfile) - # apply values reverting changes - changes.reverse() - done = set() - for action in changes: - if action[0] == 'renamed': - oldname, newname = action[1:] - newconfig[newname] = oldconfig[oldname] - done.add(newname) - elif action[0] == 'typechanged': - optname, oldtype, newvalue = action[1:] - newconfig[optname] = newvalue - done.add(optname) - for optname, optdef in newconfig.options: - if optdef.get('type') and not optname in done: - newconfig.set_option(optname, oldconfig[optname], optdict=optdef) - - -def merge_options(options, optgroup=None): - """preprocess a list of options and remove duplicates, returning a new list - (tuple actually) of options. - - Options dictionaries are copied to avoid later side-effect. Also, if - `otpgroup` argument is specified, ensure all options are in the given group. - """ - alloptions = {} - options = list(options) - for i in range(len(options)-1, -1, -1): - optname, optdict = options[i] - if optname in alloptions: - options.pop(i) - alloptions[optname].update(optdict) - else: - optdict = optdict.copy() - options[i] = (optname, optdict) - alloptions[optname] = optdict - if optgroup is not None: - alloptions[optname]['group'] = optgroup - return tuple(options) |