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 OnDiskCollection(collections.MutableSequence):
- _file_name = None
- def __init__(self, file_dir: str, file_name: str = None, should_create: bool = True):
- super(OnDiskCollection, self).__init__()
- 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._collection = list()
- self._dirty = False
- self._should_create = should_create
- self._is_loaded = False
- @guard_loaded
- def __repr__(self):
- return "<{0} {1}>".format(self.__class__.__name__, self._collection)
- @guard_loaded
- def __len__(self):
- """List length"""
- return len(self._collection)
- @guard_loaded
- def __getitem__(self, ii):
- """Get a list item"""
- return self._collection[ii]
- @guard_loaded
- def __delitem__(self, ii):
- """Delete an item"""
- del self._collection[ii]
- self._dirty = True
- @guard_loaded
- def __setitem__(self, ii, val):
- self._collection[ii] = val
- self._dirty = True
- @guard_loaded
- def __str__(self):
- return str(self._collection)
- @guard_loaded
- def __eq__(self, other):
- if isinstance(other, self.__class__):
- other = other._collection
- elif isinstance(other, collections.MutableSequence):
- pass
- else:
- return NotImplemented
- return dict(self._collection) == dict(other)
- @guard_loaded
- def insert(self, ii, val):
- self._collection.insert(ii, val)
- self._dirty = True
- def exists(self):
- return os.path.isfile(self._file_path)
- 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 collection')
- self._collection.extend(data)
- except IOError:
- with open(self._file_path, 'w+') as f:
- try:
- json.dump(self._collection, f)
- except ValueError:
- raise ValueError('Unable to serialize on-disk collection')
- self._is_loaded = True
- def save(self):
- if self._dirty:
- with open(self._file_path, 'w') as f:
- try:
- json.dump(self._collection, f)
- except ValueError:
- raise ValueError('Unable to serialize on-disk collection')
- self._dirty = False
- def __enter__(self):
- self.load()
- return self
- def __exit__(self, exc_type, exc_val, exc_tb):
- self.save()
Add Comment
Please, Sign In to add comment