Infra/tiramisu
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

work on MetaConfig

Browse Source
This commit is contained in:
Emmanuel Garette
2013-05-02 11:34:57 +02:00
parent 9ceaa4b235
commit 2d57ed160e
6 changed files with 497 additions and 254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

555
tiramisu/config.py
View File

@ -21,7 +21,7 @@
# the whole pypy projet is under MIT licence
# ____________________________________________________________
#from inspect import getmembers, ismethod
from tiramisu.error import PropertiesOptionError
from tiramisu.error import PropertiesOptionError, ConfigError
from tiramisu.option import OptionDescription, Option, SymLinkOption
from tiramisu.setting import groups, Setting
from tiramisu.value import Values
@ -30,7 +30,7 @@ from tiramisu.i18n import _
class SubConfig(object):
"sub configuration management entry"
__slots__ = ('_cfgimpl_descr', '_cfgimpl_context')
__slots__ = ('_cfgimpl_context', '_cfgimpl_descr')
def __init__(self, descr, context):
""" Configuration option management master class
@ -42,101 +42,23 @@ class SubConfig(object):
"""
# main option description
if not isinstance(descr, OptionDescription):
raise ValueError(_('descr must be an optiondescription, not {0}').format(type(descr)))
raise ValueError(_('descr must be an optiondescription, not {0}'
'').format(type(descr)))
self._cfgimpl_descr = descr
# sub option descriptions
if not isinstance(context, SubConfig):
raise ValueError('context must be a SubConfig')
self._cfgimpl_context = context
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_reset_cache(self, only_expired=False, only=('values',
'settings')):
self.cfgimpl_get_context().cfgimpl_reset_cache(only_expired, only)
def cfgimpl_get_consistancies(self):
return self.cfgimpl_get_context().cfgimpl_get_description()._consistancies
def cfgimpl_get_description(self):
return self._cfgimpl_descr
def cfgimpl_reset_cache(self, only_expired=False, only=('values', 'settings')):
self.cfgimpl_get_context().cfgimpl_reset_cache(only_expired, only)
# ____________________________________________________________
# 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)
child = getattr(self._cfgimpl_descr, name)
if not isinstance(child, SymLinkOption):
self.cfgimpl_get_values().setitem(child, value,
force_permissive=force_permissive)
else:
context = self.cfgimpl_get_context()
path = context.cfgimpl_get_description().optimpl_get_path_by_opt(child._opt)
context._setattr(path, value, force_permissive=force_permissive)
def __delattr__(self, name):
child = getattr(self._cfgimpl_descr, name)
del(self.cfgimpl_get_values()[child])
def __getattr__(self, name):
return self._getattr(name)
def _getattr(self, name, force_permissive=False, force_properties=None,
validate=True):
"""
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,
force_permissive=force_permissive,
force_properties=force_properties)
return homeconfig._getattr(name, force_permissive=force_permissive,
force_properties=force_properties,
validate=validate)
# special attributes
if name.startswith('_cfgimpl_') or name.startswith('cfgimpl_'):
# if it were in __dict__ it would have been found already
object.__getattr__(self, name)
opt_or_descr = getattr(self.cfgimpl_get_description(), name)
# symlink options
if isinstance(opt_or_descr, SymLinkOption):
context = self.cfgimpl_get_context()
path = context.cfgimpl_get_description().optimpl_get_path_by_opt(opt_or_descr._opt)
return context._getattr(path, validate=validate,
force_properties=force_properties,
force_permissive=force_permissive)
elif isinstance(opt_or_descr, OptionDescription):
self.cfgimpl_get_settings().validate_properties(opt_or_descr,
True, False,
force_permissive=force_permissive,
force_properties=force_properties)
return SubConfig(opt_or_descr, self._cfgimpl_context)
else:
return self.cfgimpl_get_values().getitem(opt_or_descr,
validate=validate,
force_properties=force_properties,
force_permissive=force_permissive)
def cfgimpl_get_home_by_path(self, path, force_permissive=False, force_properties=None):
def cfgimpl_get_home_by_path(self, path, force_permissive=False,
force_properties=None):
""":returns: tuple (config, name)"""
path = path.split('.')
for step in path[:-1]:
@ -202,7 +124,8 @@ class SubConfig(object):
if isinstance(child, OptionDescription):
try:
if group_type is None or (group_type is not None and
child.optimpl_get_group_type() == group_type):
child.optimpl_get_group_type()
== group_type):
yield child._name, getattr(self, child._name)
except GeneratorExit:
raise StopIteration
@ -224,10 +147,91 @@ class SubConfig(object):
__repr__ = __str__
def cfgimpl_get_path(self):
descr = self.cfgimpl_get_description()
context_descr = self.cfgimpl_get_context().cfgimpl_get_description()
return context_descr.optimpl_get_path_by_opt(descr)
def cfgimpl_get_context(self):
return self._cfgimpl_context
def cfgimpl_get_description(self):
if self._cfgimpl_descr is None:
raise ConfigError(_('no optiondescription for this config (may be MetaConfig without meta)'))
else:
return self._cfgimpl_descr
def cfgimpl_get_settings(self):
return self.cfgimpl_get_context()._cfgimpl_settings
def cfgimpl_get_values(self):
return self.cfgimpl_get_context()._cfgimpl_values
# ____________________________________________________________
# 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)
child = getattr(self.cfgimpl_get_description(), name)
if not isinstance(child, SymLinkOption):
self.cfgimpl_get_values().setitem(child, value,
force_permissive=force_permissive)
else:
context = self.cfgimpl_get_context()
path = context.cfgimpl_get_description().optimpl_get_path_by_opt(child._opt)
context._setattr(path, value, force_permissive=force_permissive)
def __delattr__(self, name):
child = getattr(self.cfgimpl_get_description(), name)
del(self.cfgimpl_get_values()[child])
def __getattr__(self, name):
return self._getattr(name)
def _getattr(self, name, force_permissive=False, force_properties=None,
validate=True):
"""
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,
force_permissive=force_permissive,
force_properties=force_properties)
return homeconfig._getattr(name, force_permissive=force_permissive,
force_properties=force_properties,
validate=validate)
# special attributes
if name.startswith('_cfgimpl_') or name.startswith('cfgimpl_'):
# if it were in __dict__ it would have been found already
return object.__getattribute__(self, name)
opt_or_descr = getattr(self.cfgimpl_get_description(), name)
# symlink options
if isinstance(opt_or_descr, SymLinkOption):
context = self.cfgimpl_get_context()
path = context.cfgimpl_get_description().optimpl_get_path_by_opt(opt_or_descr._opt)
return context._getattr(path, validate=validate,
force_properties=force_properties,
force_permissive=force_permissive)
elif isinstance(opt_or_descr, OptionDescription):
self.cfgimpl_get_settings().validate_properties(opt_or_descr,
True, False,
force_permissive=force_permissive,
force_properties=force_properties)
return SubConfig(opt_or_descr, self.cfgimpl_get_context())
else:
return self.cfgimpl_get_values().getitem(opt_or_descr,
validate=validate,
force_properties=force_properties,
force_permissive=force_permissive)
def find(self, bytype=None, byname=None, byvalue=None, type_='option'):
"""
@ -243,7 +247,8 @@ class SubConfig(object):
type_=type_,
_subpath=self.cfgimpl_get_path())
def find_first(self, bytype=None, byname=None, byvalue=None, type_='option'):
def find_first(self, bytype=None, byname=None, byvalue=None,
type_='option'):
"""
finds an option recursively in the config
@ -257,117 +262,6 @@ class SubConfig(object):
type_=type_,
_subpath=self.cfgimpl_get_path())
def make_dict(self, flatten=False, _currpath=None, withoption=None, withvalue=None):
"""export the whole config into a `dict`
:returns: dict of Option's name (or path) and values"""
pathsvalues = []
if _currpath is None:
_currpath = []
if withoption is None and withvalue is not None:
raise ValueError(_("make_dict can't filtering with value without option"))
if withoption is not None:
mypath = self.cfgimpl_get_path()
for path in self.cfgimpl_get_context()._find(bytype=Option,
byname=withoption,
byvalue=withvalue,
first=False,
type_='path',
_subpath=mypath):
path = '.'.join(path.split('.')[:-1])
opt = self.cfgimpl_get_context().cfgimpl_get_description().optimpl_get_opt_by_path(path)
if mypath is not None:
if mypath == path:
withoption = None
withvalue = None
break
else:
tmypath = mypath + '.'
if not path.startswith(tmypath):
raise AttributeError(_('unexpected path {0}, '
'should start with {1}').format(path, mypath))
path = path[len(tmypath):]
self._make_sub_dict(opt, path, pathsvalues, _currpath, flatten)
#withoption can be set to None below !
if withoption is None:
for opt in self.cfgimpl_get_description().optimpl_getchildren():
path = opt._name
self._make_sub_dict(opt, path, pathsvalues, _currpath, flatten)
if _currpath == []:
options = dict(pathsvalues)
return options
return pathsvalues
def _make_sub_dict(self, opt, path, pathsvalues, _currpath, flatten):
if isinstance(opt, OptionDescription):
try:
pathsvalues += getattr(self, path).make_dict(flatten,
_currpath + path.split('.'))
except PropertiesOptionError:
pass # this just a hidden or disabled option
else:
try:
value = self._getattr(opt._name)
if flatten:
name = opt._name
else:
name = '.'.join(_currpath + [opt._name])
pathsvalues.append((name, value))
except PropertiesOptionError:
pass # this just a hidden or disabled option
# ____________________________________________________________
class Config(SubConfig):
"main configuration management entry"
__slots__ = ('_cfgimpl_settings', '_cfgimpl_values')
def __init__(self, descr):
""" Configuration option management master class
:param descr: describes the configuration schema
:type descr: an instance of ``option.OptionDescription``
:param context: the current root config
:type context: `Config`
"""
self._cfgimpl_settings = Setting(self)
self._cfgimpl_values = Values(self)
super(Config, self).__init__(descr, self) # , slots)
self._cfgimpl_build_all_paths()
def _cfgimpl_build_all_paths(self):
self._cfgimpl_descr.optimpl_build_cache()
def read_only(self):
self.cfgimpl_get_settings().read_only()
def read_write(self):
self.cfgimpl_get_settings().read_write()
def getowner(self, path):
opt = self.cfgimpl_get_description().optimpl_get_opt_by_path(path)
return self.cfgimpl_get_values().getowner(opt)
def cfgimpl_reset_cache(self, only_expired=False, only=('values', 'settings')):
if 'values' in only:
self.cfgimpl_get_values().reset_cache(only_expired=only_expired)
if 'settings' in only:
self.cfgimpl_get_settings().reset_cache(only_expired=only_expired)
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 cfgimpl_get_path(self):
return None
def _find(self, bytype, byname, byvalue, first, type_='option',
_subpath=None, check_properties=True):
"""
@ -377,12 +271,12 @@ class Config(SubConfig):
: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
try:
if byname is None or path == byname or path.endswith('.' + byname):
return True
except IndexError:
pass
return False
def _filter_by_value():
if byvalue is None:
@ -413,7 +307,7 @@ class Config(SubConfig):
# else:
# return False
# return False
if type_ not in ('option', 'path', 'value'):
if type_ not in ('option', 'path', 'context', 'value'):
raise ValueError(_('unknown type_ type {0} for _find').format(type_))
find_results = []
opts, paths = self.cfgimpl_get_description()._cache_paths
@ -443,59 +337,230 @@ class Config(SubConfig):
retval = value
elif type_ == 'path':
retval = path
else:
elif type_ == 'option':
retval = option
elif type_ == 'context':
retval = self.cfgimpl_get_context()
if first:
return retval
else:
find_results.append(retval)
if find_results == []:
raise AttributeError(_("no option found in config with these criteria"))
#FIXME too slow
#raise AttributeError(_("no option found in config with these criteria"))
raise AttributeError("no option found in config with these criteria")
else:
return find_results
def make_dict(self, flatten=False, _currpath=None, withoption=None,
withvalue=None):
"""export the whole config into a `dict`
:returns: dict of Option's name (or path) and values"""
pathsvalues = []
if _currpath is None:
_currpath = []
if withoption is None and withvalue is not None:
raise ValueError(_("make_dict can't filtering with value without "
"option"))
if withoption is not None:
mypath = self.cfgimpl_get_path()
for path in self.cfgimpl_get_context()._find(bytype=Option,
byname=withoption,
byvalue=withvalue,
first=False,
type_='path',
_subpath=mypath):
path = '.'.join(path.split('.')[:-1])
opt = self.cfgimpl_get_context().cfgimpl_get_description().optimpl_get_opt_by_path(path)
if mypath is not None:
if mypath == path:
withoption = None
withvalue = None
break
else:
tmypath = mypath + '.'
if not path.startswith(tmypath):
raise AttributeError(_('unexpected path {0}, '
'should start with {1}'
'').format(path, mypath))
path = path[len(tmypath):]
self._make_sub_dict(opt, path, pathsvalues, _currpath, flatten)
#withoption can be set to None below !
if withoption is None:
for opt in self.cfgimpl_get_description().optimpl_getchildren():
path = opt._name
self._make_sub_dict(opt, path, pathsvalues, _currpath, flatten)
if _currpath == []:
options = dict(pathsvalues)
return options
return pathsvalues
class MetaConfig(object):
__slots__ = ('_children')
def _make_sub_dict(self, opt, path, pathsvalues, _currpath, flatten):
if isinstance(opt, OptionDescription):
try:
pathsvalues += getattr(self, path).make_dict(flatten,
_currpath + path.split('.'))
except PropertiesOptionError:
pass # this just a hidden or disabled option
else:
try:
value = self._getattr(opt._name)
if flatten:
name = opt._name
else:
name = '.'.join(_currpath + [opt._name])
pathsvalues.append((name, value))
except PropertiesOptionError:
pass # this just a hidden or disabled option
def __init__(self, children):
def cfgimpl_get_path(self):
descr = self.cfgimpl_get_description()
context_descr = self.cfgimpl_get_context().cfgimpl_get_description()
return context_descr.optimpl_get_path_by_opt(descr)
class ConfigCommon(SubConfig):
__slots__ = ('_cfgimpl_values', '_cfgimpl_settings', '_cfgimpl_meta')
def _cfgimpl_build_all_paths(self):
self.cfgimpl_get_description().optimpl_build_cache()
def read_only(self):
self.cfgimpl_get_settings().read_only()
def read_write(self):
self.cfgimpl_get_settings().read_write()
def getowner(self, path):
opt = self.cfgimpl_get_description().optimpl_get_opt_by_path(path)
return self.cfgimpl_get_values().getowner(opt)
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_get_description(), path)
return getattr(self.cfgimpl_get_description(), path)
def cfgimpl_get_path(self):
return None
def cfgimpl_get_meta(self):
return self._cfgimpl_meta
# ____________________________________________________________
class Config(ConfigCommon):
"main configuration management entry"
__slots__ = tuple()
def __init__(self, descr):
""" Configuration option management master class
:param descr: describes the configuration schema
:type descr: an instance of ``option.OptionDescription``
:param context: the current root config
:type context: `Config`
"""
self._cfgimpl_settings = Setting(self)
self._cfgimpl_values = Values(self)
super(Config, self).__init__(descr, self) # , slots)
self._cfgimpl_build_all_paths()
self._cfgimpl_meta = None
def cfgimpl_reset_cache(self, only_expired=False, only=('values', 'settings')):
if 'values' in only:
self.cfgimpl_get_values().reset_cache(only_expired=only_expired)
if 'settings' in only:
self.cfgimpl_get_settings().reset_cache(only_expired=only_expired)
class MetaConfig(ConfigCommon):
__slots__ = ('_cfgimpl_children',)
def __init__(self, children, meta=True):
if not isinstance(children, list):
raise ValueError(_("metaconfig's children must be a list"))
descr = None
for child in children:
if not isinstance(child, Config):
raise ValueError(_("metaconfig's children must be Config, not {0}"
"".format(type(Config))))
if descr is None:
descr = child.cfgimpl_get_description()
elif descr is child.cfgimpl_get_description():
raise ValueError(_('all config in MetaConfig must have same '
'optiondescription'))
self._cfgimpl_descr = None
if meta:
for child in children:
if not isinstance(child, ConfigCommon):
raise ValueError(_("metaconfig's children must be Config, not {0}"
"".format(type(child))))
if self._cfgimpl_descr is None:
self._cfgimpl_descr = child.cfgimpl_get_description()
elif not self._cfgimpl_descr is child.cfgimpl_get_description():
raise ValueError(_('all config in MetaConfig must have same '
'optiondescription'))
if child.cfgimpl_get_meta() is not None:
raise ValueError(_("child has already a metaconfig's"))
child._cfgimpl_meta = self
self._children = children
self._cfgimpl_children = children
self._cfgimpl_settings = Setting(self)
self._cfgimpl_values = Values(self)
self._cfgimpl_meta = None
def _find(self, bytype, byname, byvalue, first, type_):
if type_ not in ('option', 'path'):
raise ValueError(_('unknown type_ type {0} for _find'
'').format(type_))
#all children have same optiondescription, search in first one's
return self._children[0]._find(bytype, byname, byvalue, first=first,
type_=type_, check_properties=False)
def cfgimpl_get_context(self):
return self
def find(self, bytype=None, byname=None, byvalue=None, type_='option'):
return self._find(bytype, byname, byvalue, type_, first=False)
def cfgimpl_reset_cache(self, only_expired=False, only=('values', 'settings')):
if 'values' in only:
self.cfgimpl_get_values().reset_cache(only_expired=only_expired)
if 'settings' in only:
self.cfgimpl_get_settings().reset_cache(only_expired=only_expired)
for child in self._cfgimpl_children:
child.cfgimpl_reset_cache(only_expired=only_expired, only=only)
def find_first(self, bytype=None, byname=None, byvalue=None,
type_='option'):
return self._find(bytype, byname, byvalue, type_, first=True)
def __setattr__(self, name, value):
for child in self._children:
def set_contexts(self, path, value):
for child in self._cfgimpl_children:
try:
setattr(child, name, value)
if not isinstance(child, MetaConfig):
setattr(child, path, value)
else:
child.set_contexts(path, value)
except PropertiesOptionError:
pass
def find_first_contexts(self, byname=None, bypath=None, byvalue=None, type_='context'):
ret = []
try:
if bypath is None and byname is not None and \
self.cfgimpl_get_description() is not None:
bypath = self._find(bytype=None, byvalue=None, byname=byname,
first=True, type_='path',
check_properties=False)
except ConfigError:
pass
for child in self._cfgimpl_children:
try:
if not isinstance(child, MetaConfig):
if bypath is not None:
if byvalue is not None:
if getattr(child, bypath) == byvalue:
ret.append(child)
else:
#not raise
getattr(child, bypath)
ret.append(child)
else:
ret.append(child.find_first(byname=byname,
byvalue=byvalue,
type_=type_))
else:
ret.extend(child.find_first_contexts(byname=byname,
bypath=bypath,
byvalue=byvalue,
type_=type_))
except AttributeError:
pass
return ret
def mandatory_warnings(config):
"""convenience function to trace Options that are mandatory and

1
tiramisu/error.py
View File

@ -40,6 +40,7 @@ class PropertiesOptionError(AttributeError):
#____________config___________
class ConfigError(StandardError):
"""try to change owner for an option without value
or if _cfgimpl_descr is None
or if error in calculation"""
pass

8
tiramisu/option.py
View File

@ -31,9 +31,9 @@ from tiramisu.i18n import _
from tiramisu.autolib import carry_out_calculation
name_regexp = re.compile(r'^\d+')
forbidden_names = ['iter_all', 'iter_group', 'find', 'find_fisrt',
forbidden_names = ['iter_all', 'iter_group', 'find', 'find_first',
'make_dict', 'unwrap_from_path', 'read_only',
'read_write', 'getowner']
'read_write', 'getowner', 'set_contexts']
def valid_name(name):
@ -43,8 +43,8 @@ def valid_name(name):
return False
if re.match(name_regexp, name) is None and not name.startswith('_') \
and name not in forbidden_names \
and not name.startswith('optimpl_') and \
not name.startswith('cfgimpl_'):
and not name.startswith('optimpl_') \
and not name.startswith('cfgimpl_'):
return True
else:
return False

17
tiramisu/value.py
View File

@ -39,11 +39,18 @@ class Values(object):
self._cache = {}
super(Values, self).__init__()
def _get_default(self, opt):
meta = self.context.cfgimpl_get_meta()
if meta is not None:
return meta.cfgimpl_get_values()[opt]
else:
return opt.optimpl_getdefault()
def _get_value(self, opt):
"return value or default value if not set"
#if no value
if opt not in self._values:
value = opt.optimpl_getdefault()
value = self._get_default(opt)
if opt.optimpl_is_multi():
value = Multi(value, self.context, opt)
else:
@ -114,7 +121,7 @@ class Values(object):
self._reset(opt)
# frozen and force default
elif is_frozen and 'force_default_on_freeze' in setting[opt]:
value = opt.optimpl_getdefault()
value = self._get_default(opt)
if opt.optimpl_is_multi():
value = Multi(value, self.context, opt)
if validate and not opt.optimpl_validate(value, self.context, 'validator' in setting):
@ -154,7 +161,11 @@ class Values(object):
self._values[opt] = (setting.getowner(), value)
def getowner(self, opt):
return self._values.get(opt, (owners.default, None))[0]
owner = self._values.get(opt, (owners.default, None))[0]
meta = self.context.cfgimpl_get_meta()
if owner is owners.default and meta is not None:
owner = meta.cfgimpl_get_values().getowner(opt)
return owner
def setowner(self, opt, owner):
if opt not in self._values: