Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import collections
- import json
- import os
- import pathlib
- import functools
- class StoreDoesNotExistError(Exception):
- """
- Exception raised when specified store does not exist
- """
- pass
- def guard_loaded(fn):
- @functools.wraps(fn)
- def wrapper(self, *args, **kwargs):
- if not self._is_loaded:
- self.load()
- return fn(self, *args, **kwargs)
- return wrapper
- class OnDiskMapping(collections.MutableMapping):
- _file_name = None
- def __init__(self, file_dir: str, file_name: str = None, should_create: bool = True):
- self._file_dir = os.path.expanduser(file_dir)
- if file_name:
- self._file_name = file_name
- else:
- if not self._file_name:
- raise ValueError('File name for store not specified')
- self._file_path = os.path.join(self._file_dir, self._file_name)
- self._store = dict()
- self._dirty = False
- self._should_create = should_create
- self._is_loaded = True
- def load(self):
- if self.exists():
- pass
- else:
- if self._should_create:
- pathlib.Path(self._file_dir).mkdir(exist_ok=True, parents=True)
- else:
- raise StoreDoesNotExistError('Store does not exist at {}'.format(self._file_path))
- try:
- with open(self._file_path, 'r+') as f:
- try:
- data = json.load(f)
- except json.JSONDecodeError:
- raise ValueError('Unable to deserialize on-disk mapping')
- self._store.update(data)
- except IOError:
- with open(self._file_path, 'w+') as f:
- try:
- json.dump(self._store, f)
- except ValueError:
- raise ValueError('Unable to serialize on-disk mapping')
- self._is_loaded = True
- def save(self):
- if self._dirty:
- with open(self._file_path, 'w') as f:
- try:
- json.dump(self._store, f)
- except ValueError:
- raise ValueError('Unable to serialize on-disk mapping')
- self._dirty = False
- @guard_loaded
- def __setitem__(self, key, value):
- self._store[key] = value
- self._dirty = True
- @guard_loaded
- def __getitem__(self, key):
- return self._store[key]
- @guard_loaded
- def __delitem__(self, key):
- del self._store[key]
- self._dirty = True
- @guard_loaded
- def __iter__(self):
- return (key for key, value in self._store.values())
- @guard_loaded
- def __len__(self):
- return len(self._store)
- def exists(self):
- return os.path.isfile(self._file_path)
- @guard_loaded
- def clear(self):
- self._store = {}
- self._dirty = True
- def __eq__(self, other):
- if isinstance(other, self.__class__):
- other = other._store
- elif isinstance(other, collections.MutableMapping):
- pass
- else:
- return NotImplemented
- return dict(self._store) == dict(other)
- def __enter__(self):
- self.load()
- return self
- def __exit__(self, exc_type, exc_val, exc_tb):
- self.save()
- def __del__(self):
- if os.path.isfile(self._file_path):
- os.remove(self._file_path)
- del self
- @guard_loaded
- def copy(self):
- store = OnDiskMapping(self._file_dir, self._file_name, self._should_create)
- store.update(self._store)
- return store
- def __repr__(self):
- return '%s(%r)' % (self.__class__.__name__, dict(self.items()))
Add Comment
Please, Sign In to add comment