Source code for envdir.env

import os

try:
    from UserDict import IterableUserDict as UserDict
except ImportError:
    from collections import UserDict


def isenvvar(name):
    root, name = os.path.split(name)
    return '=' not in name


class _EmptyFile(Exception):
    pass


try:
    FileNotFoundError
except NameError:  # <python3
    FileNotFoundError = OSError

_sentinel = object()


[docs]class Env(UserDict): """ An dict-like object to represent an envdir environment with extensive API, can be used as context manager, too. """
[docs] def __init__(self, path): self.path = path self.data = {} self.originals = {} self.created = {} self._load()
[docs] def __repr__(self): return "<envdir.Env '%s'>" % self.path
[docs] def __enter__(self): return self
[docs] def __exit__(self, type, value, traceback): self.clear()
[docs] def __getitem__(self, name, default=_sentinel): try: return self._get(name, default=default) except (_EmptyFile, FileNotFoundError): if default is _sentinel: raise KeyError(name) return default
[docs] def __setitem__(self, name, value): self._write(**{name: value}) self._set(name, value) self.created[name] = value
[docs] def __delitem__(self, name): os.remove(os.path.join(self.path, name)) self._delete(name)
[docs] def __contains__(self, name): return (name in self.data or os.path.exists(os.path.join(self.path, name)))
def _load(self): for _, _, files in os.walk(self.path, followlinks=True): for path in filter(isenvvar, files): root, name = os.path.split(path) try: value = self._get(name) except _EmptyFile: self._delete(name) else: self._set(name, value) def _open(self, name, mode='r'): return open(os.path.join(self.path, name), mode) def _get(self, name, default=_sentinel): path = os.path.join(self.path, name) if os.stat(path).st_size == 0: raise _EmptyFile if not os.path.exists(path): return default with self._open(name) as var: return var.read().strip('\n').replace('\x00', '\n') def _set(self, name, value): if name in os.environ: self.originals[name] = os.environ[name] self.data[name] = value os.environ[name] = value def _delete(self, name): if name in self.originals: os.environ[name] = self.originals[name] elif name in os.environ: del os.environ[name] if name in self.data: del self.data[name] def _write(self, **values): for name, value in values.items(): with self._open(name, 'w') as env: env.write(value)
[docs] def clear(self): """ Clears the envdir by resetting the os.environ items to the values it had before opening this envdir (or removing them if they didn't exist). Doesn't delete the envdir files. """ for name in list(self.data.keys()): self._delete(name)