On disk dict/mapping

on_disk_mapping.py

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()))