Da

✏️ Editing: data_storage.py

import threading

from parallels.core.logging import get_logger
from parallels.core.utils.json_utils import encode_json, decode_json

logger = get_logger(__name__)

class BaseDataStorage(object):
    def get_data(self, key, default=None):
        """Retrieve data from storage

:type key: str | None
        :type default: mixed | None
        :rtype: mixed
        """
        raise NotImplementedError()

def set_data(self, key, data):
        """Place data into storage

:type key: str
        :type data: mixed
        """
        raise NotImplementedError()

class DataStorage(BaseDataStorage):
    def __init__(self):
        self._data = {}
        try:
            self._data = self._load()
        except Exception:
            logger.exception()

def get_data(self, key, default=None):
        """Retrieve data from storage

:type key: str | None
        :type default: mixed | None
        :rtype: mixed
        """
        return self._data.get(key, default)

def set_data(self, key, data):
        """Place data into storage

:type key: str
        :type data: mixed
        """
        self._data[key] = data
        try:
            self._save(self._data)
        except Exception:
            logger.exception()

def _load(self):
        raise NotImplementedError()

def _save(self, data):
        raise NotImplementedError()

class InnerDataStorage(BaseDataStorage):
    def __init__(self, data_storage, inner_storage_key):
        """
        :type data_storage: parallels.core.utils.data_storage.BaseDataStorage
        :type inner_storage_key: str | unicode
        """
        self._data_storage = data_storage
        self._inner_storage_key = inner_storage_key

def get_data(self, key, default=None):
        """Retrieve data from storage

:type key: str | None
        :type default: mixed | None
        :rtype: mixed
        """
        storage_data = self._data_storage.get_data(self._inner_storage_key)
        if storage_data is None:
            return default
        else:
            return storage_data.get(key, default)

def set_data(self, key, data):
        """Place data into storage

:type key: str
        :type data: mixed
        """
        storage_data = self._data_storage.get_data(self._inner_storage_key)
        if storage_data is None:
            storage_data = {}

storage_data[key] = data
        self._data_storage.set_data(self._inner_storage_key, storage_data)

class SessionFileDataStorage(DataStorage):
    def __init__(self, session_file):
        """
        :type session_file: parallels.core.session_files.SessionFile
        """
        self._session_file = session_file
        self._lock = threading.Lock()
        super(SessionFileDataStorage, self).__init__()

def _load(self):
        with self._lock:
            if not self._session_file.is_exists():
                return {}
            return decode_json(self._session_file.read())

def _save(self, data):
        with self._lock:
            self._session_file.write(encode_json(data, pretty_print=True).encode('utf-8'))

📁 Sid Gifari File Manager