Source code for benchbuild.utils.dict

"""An extensible dictionary."""
from contextlib import contextmanager


[docs]def extend_as_list(original_dict, **kwargs): new_dict = original_dict for k, v in kwargs.items(): if k in original_dict: oldv = new_dict[k] if not hasattr(oldv, 'extend'): oldv = [oldv] if isinstance(v, str): v = [v] if hasattr(v, '__iter__'): oldv.extend(v) else: oldv.append(v) new_dict[k] = oldv else: new_dict[k] = v return new_dict
[docs]class ExtensibleDict: """A dictionary that provides temporary modification.""" _current = dict() _extender_fn = None def __init__(self, extender_fn=None): self._extender_fn = extender_fn super(ExtensibleDict, self).__init__() @contextmanager def __call__(self, *args, extender_fn=None, **kwargs): """ A context manager to temporarily modify the content of dict. Any time you enter the context the existing dictionary is updated with the content of **kwargs. When the context exits the original dictionary is restored. If ``extender_fn`` is not None the existing content of the dictionary is not overwritten, but handed to the extender_fn as first argument in addition to the changes supplied in the ``kwargs``. The result will be stored temporarily int he dictionary. Args: extender_fn: Returns: An updated dictionary. """ previous = self._current.copy() if extender_fn is None: extender_fn = self._extender_fn self.update(extender_fn, **kwargs) try: yield finally: self._current = previous def __iter__(self): return iter(self._current.items()) def __len__(self): return len(self._current) def __contains__(self, name): return name in self._current def __getitem__(self, name): return self._current[name]
[docs] def keys(self): return self._current.keys()
[docs] def values(self): return self._current.values()
[docs] def items(self): return self._current.items()
[docs] def get(self, name, *default): return self._current.get(name, *default)
def __delitem__(self, name): del self._current[name] def __setitem__(self, name, value): self._current[name] = value
[docs] def pop(self, name, *default): return self._current.pop(name, *default)
[docs] def clear(self): self._current.clear()
[docs] def update(self, extender_fn, *args, **kwargs): if extender_fn is not None: self._current.update(*args, **extender_fn(self._current, **kwargs)) else: self._current.update(*args, **kwargs)
[docs] def getdict(self): return dict((k, str(v)) for k, v in self._current.items())
def __str__(self): return str(self._current) def __repr__(self): return repr(self._current)