Sheerka-Old/sdp/sheerkaDataProviderIO.py

import io
from os import path
import os
from fs.memoryfs import MemoryFS
import logging


class SheerkaDataProviderIO:

    def __init__(self, root):
        self.root = root

    def exists(self, file_path):
        pass

    def open(self, file_path, mode):
        pass

    def read_text(self, file_path):
        pass

    def read_binary(self, file_path):
        pass

    def write_text(self, file_path, content):
        pass

    def write_binary(self, file_path, content):
        pass

    def remove(self, file_path):
        pass

    @staticmethod
    def get(root):
        if root == "mem://":
            return SheerkaDataProviderDictionaryIO()
        else:
            return SheerkaDataProviderFileIO(root)

    def get_obj_path(self, object_type, digest):
        return path.join(self.root, object_type, digest[:24], digest)

    def path_join(self, *paths):
        return path.join(self.root, *paths)


class SheerkaDataProviderFileIO(SheerkaDataProviderIO):
    log = logging.getLogger("FileIO")

    def __init__(self, root):
        root = path.abspath(path.join(path.expanduser("~"), ".sheerka")) \
            if root is None \
            else path.abspath(root)
        super().__init__(root)

        self.log.debug("root is set to '" + self.root + "'")

        if not path.exists(self.root):
            self.log.debug("root folder not found. Creating it.")
            os.makedirs(self.root)
            self.first_time = True
        else:
            self.first_time = False

    def open(self, file_path, mode):
        return open(file_path, mode)

    def read_text(self, file_path):
        with open(file_path) as f:
            return f.read()

    def read_binary(self, file_path):
        with open(file_path, "rb") as f:
            return f.read()

    def write_text(self, file_path, content):
        self._write(file_path, content, "w")

    def write_binary(self, file_path, content):
        self._write(file_path, content, "wb")

    def exists(self, file_path):
        return path.exists(file_path)

    def remove(self, file_path):
        os.remove(file_path)

    @staticmethod
    def _write(file_path, content, mode):
        if not path.exists(path.dirname(file_path)):
            os.makedirs(path.dirname(file_path))

        with open(file_path, mode) as f:
            f.write(content)


class SheerkaDataProviderMemoryIO(SheerkaDataProviderIO):
    log = logging.getLogger("MemoryIO")

    def __init__(self):
        super().__init__("")
        self.mem_fs = MemoryFS()
        self.log.debug("Initializing memory file.")
        self.first_time = True

    def open(self, file_path, mode):
        return self.mem_fs.open(file_path, mode)

    def exists(self, file_path):
        return self.mem_fs.exists(file_path)

    def read_text(self, file_path):
        return self.mem_fs.readtext(file_path)

    def read_binary(self, file_path):
        return self.mem_fs.readbytes(file_path)

    def write_binary(self, file_path, content):
        self._ensure_parent_folder(file_path)
        self.mem_fs.writebytes(file_path, content)

    def write_text(self, file_path, content):
        self._ensure_parent_folder(file_path)
        self.mem_fs.writetext(file_path, content)

    def remove(self, file_path):
        self.mem_fs.remove(file_path)

    def _ensure_parent_folder(self, file_path):
        if not self.mem_fs.exists(path.dirname(file_path)):
            self.mem_fs.makedirs(path.dirname(file_path))


class SheerkaDataProviderDictionaryIO(SheerkaDataProviderIO):
    log = logging.getLogger("DictionaryIO")

    def __init__(self):
        super().__init__("")
        self.cache = {}
        self.log.debug("Initializing dictionary file.")
        self.first_time = True

    def exists(self, file_path):
        if file_path == "":
            return True

        return file_path in self.cache

    def read_text(self, file_path):
        return self.cache[file_path]

    def read_binary(self, file_path):
        return self.cache[file_path]

    def write_binary(self, file_path, content):
        self.cache[file_path] = content

    def write_text(self, file_path, content):
        self.cache[file_path] = content

    def remove(self, file_path):
        del (self.cache[file_path])

    def open(self, file_path, mode):
        if "w" in mode:
            stream = io.BytesIO() if "b" in mode else io.StringIO()
            stream.close = on_close(self, file_path, stream)(stream.close)
            return stream

        return io.BytesIO(self.cache[file_path]) if "b" in mode else io.StringIO(self.cache[file_path])


def on_close(dictionary_io, file_path, stream):
    """
    Decorator to intercept the close.
    I guess that there are solution that are more elegant
    :param dictionary_io:
    :param file_path:
    :param stream:
    :return:
    """
    def decorator(func):
        def wrapper(*args, **kwargs):
            stream.seek(0)
            dictionary_io.cache[file_path] = stream.read()
            func(*args, **kwargs)
        return wrapper
    return decorator