On disk list/collection

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