Source code for layeredconfig.layeredconfig

# -*- coding: utf-8 -*-

import itertools
import logging
from datetime import datetime, date

    from collections import OrderedDict
except ImportError:  # pragma: no cover
    # if on python 2.6
    from ordereddict import OrderedDict

[docs]class LayeredConfig(object): def __init__(self, *sources, **kwargs): """Creates a config object from one or more sources and provides unified access to a nested set of configuration parameters. The source of these parameters a config file (using .ini-file style syntax), command line parameters, and default settings embedded in code. Command line parameters override configuration file parameters, which in turn override default settings in code (hence **Layered** Config). Configuration parameters are accessed as regular object attributes, not dict-style key/value pairs. Configuration parameter names should therefore be regular python identifiers, and preferrably avoid upper-case and "_" as well (i.e. only consist of the characters a-z and 0-9) Configuration parameter values can be typed (strings, integers, booleans, dates, lists...). Even though some sources lack typing information (eg in INI files, command-line parameters and enviroment variables, everything is a string), LayeredConfig will attempt to find typing information in other sources and convert data. :param \*sources: Initialized ConfigSource-derived objects :param cascade: If an attempt to get a non-existing parameter on a sub (nested) configuration object should attempt to get the parameter on the parent config object. ``False`` by default, :type cascade: bool :param writable: Whether configuration values should be mutable. ``True`` by default. This does not affect :py:meth:`~Layeredconfig.set`. :type writable: bool """ self._sources = sources self._subsections = OrderedDict() self._cascade = kwargs.get('cascade', False) self._writable = kwargs.get('writable', True) self._parent = None self._sectionkey = None # Each source may have any number of named subsections. We # create a LayeredConfig object for each name, and stuff all # matching subections from each of our sources in it. # # 1. find all names sectionkeys = [] for src in self._sources: try: for k in src.subsections(): if k not in sectionkeys: sectionkeys.append(k) except AttributeError: # possibly others, or all # we couldn't get any subsections for source, perhaps # because it's an "empty" source. Well, that's ok. pass for k in sectionkeys: # 2. find all subsections in all of our sources s = [] for src in self._sources: if k in list(src.subsections()): s.append(src.subsection(k)) else: # create an "empty" subsection object. It's # important that all the LayeredConfig objects in a # tree have the exact same set of # ConfigSource-derived types. # print("creating empty %s" % src.__class__.__name__) s.append(src.__class__(parent=src, identifier=src.identifier, writable=src.writable, empty=True, cascade=self._cascade)) # 3. create a LayeredConfig object for the subsection c = self.__class__(*s, cascade=self._cascade, writable=self._writable) c._sectionkey = k c._parent = self self._subsections[k] = c # 4. give each source a chance to to some post-init setup. for src in self._sources: src.setup(self)
[docs] @staticmethod def write(config): """Commits any pending modifications, ie save a configuration file if it has been marked "dirty" as a result of an normal assignment. The modifications are written to the first writable source in this config object. .. note:: This is a static method, ie not a method on any object instance. This is because all attribute access on a LayeredConfig object is meant to retrieve configuration settings. :param config: The configuration object to save :type config: layeredconfig.LayeredConfig """ root = config while root._parent: root = root._parent for source in root._sources: if source.writable and source.dirty:
[docs] @staticmethod def set(config, key, value, sourceid="defaults"): """Sets a value in this config object *without* marking any source dirty, and with exact control of exactly where to set the value. This is mostly useful for low-level trickery with config objects. :param config: The configuration object to set values on :param key: The parameter name :param value: The new value :param sourceid: The identifier for the underlying source that the value should be set on. """ for source in config._sources: if source.identifier == sourceid: source.set(key, value)
# What if no source is found? We silently ignore...
[docs] @staticmethod def get(config, key, default=None): """Gets a value from the config object, or return a default value if the parameter does not exist, like :py:meth:`dict.get` does. """ if hasattr(config, key): return getattr(config, key) else: return default
[docs] @staticmethod def dump(config): """Returns the entire content of the config object in a way that can be easily examined, compared or dumped to a string or file. :param config: The configuration object to dump :rtype: dict """ def _dump(element): if not isinstance(element, config.__class__): return element section = dict() for key, subsection in element._subsections.items(): section[key] = _dump(subsection) for key in element: section[key] = getattr(element, key) return section return _dump(config)
# These are methods i'd like to implement next # # @staticmethod # def where(config, key): # """returns the identifier of a source where a given key is found, or None.""" # pass # # @staticmethod # def load(config, d): # """Recreates a dump()ed config object.""" # pass
[docs] @staticmethod def datetimeconvert(value): """Convert the string *value* to a :py:class:`~datetime.datetime` object. *value* is assumed to be on the form "YYYY-MM-DD HH:MM:SS" (optionally ending with fractions of a second). """ try: return datetime.strptime(value, "%Y-%m-%d %H:%M:%S.%f") except ValueError: return datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
[docs] @staticmethod def dateconvert(value): """Convert the string *value* to a :py:class:`` object. *value* is assumed to be on the form "YYYY-MM-DD". """ return datetime.strptime(value, "%Y-%m-%d").date()
[docs] @staticmethod def boolconvert(value): """Convert the string *value* to a boolean. ``"True"`` is converted to ``True`` and ``"False"`` is converted to ``False``. .. note:: If value is neither "True" nor "False", it's returned unchanged. """ # not all bools should be converted, see test_typed_commandline if value == "True": return True elif value == "False": return False else: return value
def __repr__(self): return self.dump(self).__repr__() def __iter__(self): l = set() iterables = [x.keys() for x in self._sources] if self._cascade: c = self while c._parent: iterables.append(c._parent) c = c._parent for k in itertools.chain(*iterables): if k not in l: l.add(k) yield k def __getattr__(self, name): if name in self._subsections: return self._subsections[name] found = False # find the appropriate value in the highest-priority source for source in reversed(self._sources): # if self._cascade, we must climb the entire chain of # .parent objects to be sure. done = False while not done: if source.has(name): found = True done = True # we found it elif self._cascade and source.parent: source = source.parent else: done = True # we didn't find it if found: break if found: if source.typed(name): return source.get(name) else: # we need to find a typesource for this value. done = False this = self while not done: for typesource in reversed(this._sources): if typesource.typed(name): done = True break if not done and self._cascade and this._parent: # Iterate up the parent chain to find it. this = this._parent else: done = True if typesource.typed(name): return typesource.typevalue(name, source.get(name)) else: # we can't type this data, return as-is return source.get(name) else: if self._cascade and self._parent and name not in self._parent._subsections: return self._parent.__getattr__(name) raise AttributeError("Configuration key %s doesn't exist" % name) def __setattr__(self, name, value): # print("__setattribute__ %s to %s" % (name,value)) if name.startswith("_"): object.__setattr__(self, name, value) return # we need to get access to two sources: # 1. the highest-priority writable source (regardless of # whether it originally had this value) found = False for writesource in reversed(self._sources): if writesource.writable: found = True break if found: writesource.set(name, value) writesource.dirty = True while writesource.parent: writesource = writesource.parent writesource.dirty = True # 2. the highest-priority source that has this value (typed or # not) or contains typing info for it. found = False for source in reversed(self._sources): if source.has(name) or source.typed(name): found = True break if found: source.set(name, value) # regardless of typing elif self._cascade and self._parent: return self._parent.__setattr__(name, value) else: raise AttributeError("Configuration key %s doesn't exist" % name)