Skip to content

Instantly share code, notes, and snippets.

@mfdeux
Created February 13, 2018 16:13
Show Gist options
  • Save mfdeux/5d44666d352442f24ecfbcc4e17614bc to your computer and use it in GitHub Desktop.
Save mfdeux/5d44666d352442f24ecfbcc4e17614bc to your computer and use it in GitHub Desktop.
On disk dict/mapping
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()))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment