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)