# -*- coding: utf-8 -*- "pretty small and local configuration management tool" # Copyright (C) 2012-2013 Team tiramisu (see AUTHORS for all contributors) # # This program is free software; you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation; either version 2 of the License, or # (at your option) any later version. # # This program 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 General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA # # The original `Config` design model is unproudly borrowed from # the rough pypy's guys: http://codespeak.net/svn/pypy/dist/pypy/config/ # the whole pypy projet is under MIT licence # ____________________________________________________________ #from inspect import getmembers, ismethod from tiramisu.error import (PropertiesOptionError, NotFoundError, AmbigousOptionError, NoMatchingOptionFound, MandatoryError) from tiramisu.option import (OptionDescription, Option, SymLinkOption, apply_requires) from tiramisu.setting import groups, Setting from tiramisu.value import Values class SubConfig(object): "sub configuration management entry" __slots__ = ('_cfgimpl_descr', '_cfgimpl_subconfigs', '_cfgimpl_parent', '_cfgimpl_context') def __init__(self, descr, parent, context): # FIXME , slots): """ Configuration option management master class :param descr: describes the configuration schema :type descr: an instance of ``option.OptionDescription`` :param parent: parent's `Config` :type parent: `Config` :param context: the current root config :type context: `Config` """ # main option description self._cfgimpl_descr = descr # sub option descriptions self._cfgimpl_subconfigs = None self._cfgimpl_parent = parent self._cfgimpl_context = context #self._cfgimpl_build(slots) def cfgimpl_get_context(self): return self._cfgimpl_context def cfgimpl_get_settings(self): return self._cfgimpl_context._cfgimpl_settings def cfgimpl_get_values(self): return self._cfgimpl_context._cfgimpl_values def cfgimpl_get_description(self): return self._cfgimpl_descr # ____________________________________________________________ # attribute methods def __setattr__(self, name, value): "attribute notation mechanism for the setting of the value of an option" if name.startswith('_cfgimpl_'): #self.__dict__[name] = value object.__setattr__(self, name, value) return self._setattr(name, value) def _setattr(self, name, value, force_permissive=False): if '.' in name: homeconfig, name = self.cfgimpl_get_home_by_path(name) return homeconfig.__setattr__(name, value) if type(getattr(self._cfgimpl_descr, name)) != SymLinkOption: self._validate(name, getattr(self._cfgimpl_descr, name), force_permissive=force_permissive) self.setoption(name, value) def _validate(self, name, opt_or_descr, force_permissive=False): "validation for the setattr and the getattr" apply_requires(opt_or_descr, self) if not isinstance(opt_or_descr, Option) and \ not isinstance(opt_or_descr, OptionDescription): raise TypeError('Unexpected object: {0}'.format(repr(opt_or_descr))) properties = set(self.cfgimpl_get_settings().get_properties(opt_or_descr)) #remove this properties, those properties are validate in value/setting properties = properties - set(['mandatory', 'frozen']) set_properties = set(self.cfgimpl_get_settings().get_properties()) properties = properties & set_properties if force_permissive is True or self.cfgimpl_get_settings().has_property('permissive'): properties = properties - set(self.cfgimpl_get_settings().get_permissive()) properties = properties - set(self.cfgimpl_get_settings().get_permissive(self.cfgimpl_get_description())) properties = list(properties) if properties != []: raise PropertiesOptionError("trying to access" " to an option named: {0} with properties" " {1}".format(name, str(properties)), properties) def __getattr__(self, name): return self._getattr(name) def _getattr(self, name, force_permissive=False, force_properties=None): """ attribute notation mechanism for accessing the value of an option :param name: attribute name :return: option's value if name is an option name, OptionDescription otherwise """ # attribute access by passing a path, # for instance getattr(self, "creole.general.family.adresse_ip_eth0") if '.' in name: homeconfig, name = self.cfgimpl_get_home_by_path(name) return homeconfig._getattr(name) opt_or_descr = getattr(self._cfgimpl_descr, name) # symlink options if type(opt_or_descr) == SymLinkOption: rootconfig = self.cfgimpl_get_context() path = rootconfig.cfgimpl_get_description().get_path_by_opt(opt_or_descr.opt) return getattr(rootconfig, path) self._validate(name, opt_or_descr, force_permissive=force_permissive) if isinstance(opt_or_descr, OptionDescription): children = self.cfgimpl_get_description()._children if opt_or_descr not in children[1]: raise AttributeError("{0} with name {1} object has " "no attribute {2}".format(self.__class__, opt_or_descr._name, name)) return SubConfig(opt_or_descr, self, self._cfgimpl_context) # special attributes if name.startswith('_cfgimpl_'): # if it were in __dict__ it would have been found already object.__getattr__(self, name) return self.cfgimpl_get_values()._getitem(opt_or_descr, force_properties=force_properties) def setoption(self, name, value, who=None): """effectively modifies the value of an Option() (typically called by the __setattr__) """ child = getattr(self._cfgimpl_descr, name) child.setoption(self, value) def cfgimpl_get_home_by_path(self, path): """:returns: tuple (config, name)""" path = path.split('.') for step in path[:-1]: self = getattr(self, step) return self, path[-1] def _cfgimpl_get_path(self): "the path in the attribute access meaning." #FIXME optimisation subpath = [] obj = self while obj._cfgimpl_parent is not None: subpath.insert(0, obj._cfgimpl_descr._name) obj = obj._cfgimpl_parent return ".".join(subpath) def getkey(self): return self._cfgimpl_descr.getkey(self) def __hash__(self): return hash(self.getkey()) def __eq__(self, other): "Config comparison" if not isinstance(other, Config): return False return self.getkey() == other.getkey() def __ne__(self, other): "Config comparison" return not self == other # ______________________________________________________________________ def __iter__(self): """Pythonesque way of parsing group's ordered options. iteration only on Options (not OptionDescriptions)""" for child in self._cfgimpl_descr._children[1]: if not isinstance(child, OptionDescription): try: yield child._name, getattr(self, child._name) except GeneratorExit: raise StopIteration except: pass # option with properties def iter_all(self): """A way of parsing options **and** groups. iteration on Options and OptionDescriptions.""" for child in self._cfgimpl_descr._children[1]: try: yield child._name, getattr(self, child._name) except GeneratorExit: raise StopIteration except: pass # option with properties def iter_groups(self, group_type=None): """iteration on groups objects only. All groups are returned if `group_type` is `None`, otherwise the groups can be filtered by categories (families, or whatever). :param group_type: if defined, is an instance of `groups.GroupType` or `groups.MasterGroupType` that lives in `setting.groups` """ if group_type is not None: if not isinstance(group_type, groups.GroupType): raise TypeError("Unknown group_type: {0}".format(group_type)) for child in self._cfgimpl_descr._children[1]: if isinstance(child, OptionDescription): try: if group_type is not None: if child.get_group_type() == group_type: yield child._name, getattr(self, child._name) else: yield child._name, getattr(self, child._name) except GeneratorExit: raise StopIteration except: pass # ______________________________________________________________________ def cfgimpl_set_permissive(self, permissive): if not isinstance(permissive, list): raise TypeError('permissive must be a list') self.cfgimpl_get_settings().set_permissive(permissive, self.cfgimpl_get_description()) # ______________________________________________________________________ def __str__(self): "Config's string representation" lines = [] for name, grp in self.iter_groups(): lines.append("[%s]" % name) for name, value in self: try: lines.append("%s = %s" % (name, value)) except: pass return '\n'.join(lines) __repr__ = __str__ def getpaths(self, include_groups=False, allpaths=False, mandatory=False): """returns a list of all paths in self, recursively, taking care of the context of properties (hidden/disabled) :param include_groups: if true, OptionDescription are included :param allpaths: all the options (event the properties protected ones) :param mandatory: includes the mandatory options :returns: list of all paths """ paths = [] for path in self._cfgimpl_descr.getpaths(include_groups=include_groups): if allpaths: paths.append(path) else: try: getattr(self, path) except MandatoryError: if mandatory: paths.append(path) except PropertiesOptionError: pass else: paths.append(path) return paths def getpath(self): descr = self.cfgimpl_get_description() context_descr = self.cfgimpl_get_context().cfgimpl_get_description() return context_descr.get_path_by_opt(descr) def get(self, name): path = self.getpath() return self.cfgimpl_get_context().get(name, _subpath=path) def find(self, bytype=None, byname=None, byvalue=None, byattrs=None): path = self.getpath() return self.cfgimpl_get_context().find(bytype=bytype, byname=byname, byvalue=byvalue, byattrs=byattrs, _subpath=path) def find_first(self, bytype=None, byname=None, byvalue=None, byattrs=None): path = self.getpath() return self.cfgimpl_get_context().find_first(bytype=bytype, byname=byname, byvalue=byvalue, byattrs=byattrs, _subpath=path) # ____________________________________________________________ class Config(SubConfig): "main configuration management entry" __slots__ = ('_cfgimpl_settings', '_cfgimpl_values') def __init__(self, descr, valid_opt_names=True): """ Configuration option management master class :param descr: describes the configuration schema :type descr: an instance of ``option.OptionDescription`` :param parent: is None if the ``Config`` is root parent Config otherwise :type parent: ``Config`` :param context: the current root config :type context: `Config` """ self._cfgimpl_settings = Setting() self._cfgimpl_values = Values(self) #if valid_opt_names: # # some api members shall not be used as option's names ! # #FIXME fait une boucle infini ... # #methods = getmembers(self, ismethod) # #slots = tuple([key for key, value in methods # # if not key.startswith("_")]) # slots = [] #else: # slots = [] super(Config, self).__init__(descr, None, self) # , slots) self._cfgimpl_build_all_paths() def _cfgimpl_build_all_paths(self): self._cfgimpl_descr.build_cache() def unwrap_from_path(self, path): """convenience method to extract and Option() object from the Config() and it is **fast**: finds the option directly in the appropriate namespace :returns: Option() """ if '.' in path: homeconfig, path = self.cfgimpl_get_home_by_path(path) return getattr(homeconfig._cfgimpl_descr, path) return getattr(self._cfgimpl_descr, path) def set(self, **kwargs): """ do what I mean"-interface to option setting. Searches all paths starting from that config for matches of the optional arguments and sets the found option if the match is not ambiguous. :param kwargs: dict of name strings to values. """ all_paths = [p.split(".") for p in self.getpaths(allpaths=True)] for key, value in kwargs.iteritems(): key_p = key.split('.') candidates = [p for p in all_paths if p[-len(key_p):] == key_p] if len(candidates) == 1: name = '.'.join(candidates[0]) homeconfig, name = self.cfgimpl_get_home_by_path(name) try: getattr(homeconfig, name) except MandatoryError: pass except Exception, e: raise e # HiddenOptionError or DisabledOptionError homeconfig.setoption(name, value) elif len(candidates) > 1: raise AmbigousOptionError( 'more than one option that ends with %s' % (key, )) else: raise NoMatchingOptionFound( 'there is no option that matches %s' ' or the option is hidden or disabled' % (key, )) def get(self, name, _subpath=None): """ same as a `find_first()` method in a config that has identical names: it returns the first item of an option named `name` much like the attribute access way, except that the search for the option is performed recursively in the whole configuration tree. :returns: option value. """ return self._find(byname=name, bytype=None, byvalue=None, byattrs=None, first=True, getvalue=True, _subpath=_subpath) def _find(self, bytype, byname, byvalue, byattrs, first, getvalue=False, _subpath=None): """ convenience method for finding an option that lives only in the subtree :param first: return only one option if True, a list otherwise :return: find list or an exception if nothing has been found """ def _filter_by_name(): if byname is None: return True if path == byname or path.endswith('.' + byname): return True else: return False def _filter_by_value(): if byvalue is None: return True try: value = getattr(self, path) if value == byvalue: return True except: # a property restricts the access of the value pass return False def _filter_by_type(): if bytype is None: return True if isinstance(option, bytype): return True return False def _filter_by_attrs(): if byattrs is None: return True for key, value in byattrs.items(): if not hasattr(option, key): return False else: if getattr(option, key) != value: return False else: continue return True find_results = [] opts, paths = self.cfgimpl_get_description()._cache_paths for index in range(0, len(paths)): path = paths[index] option = opts[index] if isinstance(option, OptionDescription): continue if _subpath is not None and not path.startswith(_subpath + '.'): continue if not _filter_by_name(): continue if not _filter_by_value(): continue if not _filter_by_type(): continue if not _filter_by_attrs(): continue #remove option with propertyerror, ... try: value = getattr(self, path) except: # a property restricts the access of the value continue if first: if getvalue: return value else: return option else: if getvalue: find_results.append(value) else: find_results.append(option) if find_results == []: raise NotFoundError("no option found in config with these criteria") else: return find_results def find(self, bytype=None, byname=None, byvalue=None, byattrs=None, _subpath=None): """ finds a list of options recursively in the config :param bytype: Option class (BoolOption, StrOption, ...) :param byname: filter by Option._name :param byvalue: filter by the option's value :param byattrs: dict of option attributes (default, callback...) :returns: list of matching Option objects """ return self._find(bytype, byname, byvalue, byattrs, first=False, _subpath=_subpath) def find_first(self, bytype=None, byname=None, byvalue=None, byattrs=None, _subpath=None): """ finds an option recursively in the config :param bytype: Option class (BoolOption, StrOption, ...) :param byname: filter by Option._name :param byvalue: filter by the option's value :param byattrs: dict of option attributes (default, callback...) :returns: list of matching Option objects """ return self._find(bytype, byname, byvalue, byattrs, first=True, _subpath=_subpath) def make_dict(config, flatten=False): """export the whole config into a `dict` :returns: dict of Option's name (or path) and values""" paths = config.getpaths() pathsvalues = [] for path in paths: if flatten: pathname = path.split('.')[-1] else: pathname = path try: value = getattr(config, path) pathsvalues.append((pathname, value)) except: pass # this just a hidden or disabled option options = dict(pathsvalues) return options def mandatory_warnings(config): """convenience function to trace Options that are mandatory and where no value has been set :returns: generator of mandatory Option's path """ for path in config.cfgimpl_get_description().getpaths(include_groups=True): try: config._getattr(path, force_properties=('mandatory',)) except MandatoryError: yield path except PropertiesOptionError: pass