Skip to content

FilesystemStorage

Bases: StorageIOBase, FullyFeaturedMixin

Implementation of storage class which operates on filesystem items to preserve saved items between sessions. Loading procedures can optionally use caching, they do by default, therefore without disabling it you can't rely on loads being always instantly up to date with stores.

Encoding used to read text files, as well as cache can be changed using .configure() method.

Parameters:

Name Type Description Default
__root str | Path

root dir for fs storage, if __root points to file, parent directory of this file will be used.

 required
subdir Optional[str], optional

nested directory to use for file storage, when None, data will be stored directly in __root, by default "data". When __root is file, subdirectory in __root parent directory will be used.

 'data'

Example#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
>>> tmp = getfixture('tmp_path')
>>> from magic_storage import StoreType
>>> fs = FilesystemStorage(tmp)
>>> example_item = {"foo": 32}
>>> UID = "EXAMPLE UID"
>>> fs.store_as(StoreType.JSON, uid=UID, item=example_item)
'4c9e95de851b875493ba6c6dfb16b6aaae5c3e167aef9ab6edfeb0dbca2f6574'
>>> fs.is_available(UID)
True
>>> fs.load_as(StoreType.JSON, uid=UID)
{'foo': 32}
>>>
Source code in magic_storage/impl/_filesystem.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
class FilesystemStorage(StorageIOBase, FullyFeaturedMixin):
    """Implementation of storage class which operates on filesystem items to
    preserve saved items between sessions. Loading procedures can optionally
    use caching, they do by default, therefore without disabling it you can't
    rely on loads being always instantly up to date with stores.

    Encoding used to read text files, as well as cache can be changed
    using .configure() method.

    Parameters
    ----------
    __root : str | Path
        root dir for fs storage, if __root points to file, parent directory of
        this file will be used.
    subdir : Optional[str], optional
        nested directory to use for file storage, when None, data will be stored
        directly in __root, by default "data". When __root is file, subdirectory
        in __root parent directory will be used.

    Example
    -------
    ```
    >>> tmp = getfixture('tmp_path')
    >>> from magic_storage import StoreType
    >>> fs = FilesystemStorage(tmp)
    >>> example_item = {"foo": 32}
    >>> UID = "EXAMPLE UID"
    >>> fs.store_as(StoreType.JSON, uid=UID, item=example_item)
    '4c9e95de851b875493ba6c6dfb16b6aaae5c3e167aef9ab6edfeb0dbca2f6574'
    >>> fs.is_available(UID)
    True
    >>> fs.load_as(StoreType.JSON, uid=UID)
    {'foo': 32}
    >>>
    ```
    """

    def __init__(
        self, __root: str | Path, *, subdir: Optional[str] = "data"
    ) -> None:

        # store details about location in filesystem
        __root = Path(__root)
        # when __file__ is used, replace it with parent dir
        if __root.is_file():
            __root = __root.parent
        if subdir is not None:
            self._data_dir = __root / subdir
        else:
            self._data_dir = __root
        self._data_dir.mkdir(0o777, True, True)

        self._cache: Optional[RRCache] = RRCache(maxsize=128)
        self._encoding = "utf-8"
        super().__init__()

    def _filepath(self, __uid: str) -> Path:
        __uid = make_uid(__uid)
        assert isinstance(__uid, str)

        return self._data_dir / __uid

    def _get_cache(self) -> Optional[Cache]:
        return self._cache

    def _is_available(self, __uid: str) -> bool:
        fname = self._filepath(__uid)
        return fname.exists() and fname.is_file()

    @cachedmethod(_get_cache)
    def _read_text(self, uid: str) -> str:
        with AtomicFile(self._filepath(uid)) as file:
            return file.read_text(encoding=self._encoding)

    @cachedmethod(_get_cache)
    def _read_bytes(self, uid: str) -> bytes:
        with AtomicFile(self._filepath(uid)) as file:
            return file.read_bytes()

    def _write_text(self, uid: str, item: str) -> None:
        with AtomicFile(self._filepath(uid)) as file:
            file.write_text(item, encoding=self._encoding)

    def _write_bytes(self, uid: str, item: bytes) -> None:
        with AtomicFile(self._filepath(uid)) as file:
            file.write_bytes(item)

    def _delete(self, __uid: str, /, *, missing_ok: bool = False) -> None:
        self._filepath(__uid).unlink(missing_ok)

    def configure(
        self,
        *,
        encoding: str | sentinel = sentinel,
        cache: Optional[Cache] | sentinel = sentinel,
    ) -> None:
        """Configure FileStorage instance.

        Parameters
        ----------
        encoding : str | sentinel, optional
            Change encoding used to read/write text, when sentinel, old value is kept, by default "utf-8"
        cache : Optional[Cache] | sentinel, optional
            Change cache instance used for caching, set to None to disable caching, when sentinel, old value is kept, by default RRCache(maxsize=128)
        """
        if encoding is not sentinel:
            self._encoding = encoding
            logging.debug(f"Changed encoding of FileStorage to {encoding}.")

        if cache is not sentinel:
            self._cache = cache  # type: ignore
            logging.debug(f"Changed cache of FileStorage to {cache}.")

configure(*, encoding=sentinel, cache=sentinel) #

Configure FileStorage instance.

Parameters:

Name Type Description Default
encoding str | sentinel, optional

Change encoding used to read/write text, when sentinel, old value is kept, by default "utf-8"

 sentinel
cache Optional[Cache] | sentinel, optional

Change cache instance used for caching, set to None to disable caching, when sentinel, old value is kept, by default RRCache(maxsize=128)

 sentinel
Source code in magic_storage/impl/_filesystem.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
def configure(
    self,
    *,
    encoding: str | sentinel = sentinel,
    cache: Optional[Cache] | sentinel = sentinel,
) -> None:
    """Configure FileStorage instance.

    Parameters
    ----------
    encoding : str | sentinel, optional
        Change encoding used to read/write text, when sentinel, old value is kept, by default "utf-8"
    cache : Optional[Cache] | sentinel, optional
        Change cache instance used for caching, set to None to disable caching, when sentinel, old value is kept, by default RRCache(maxsize=128)
    """
    if encoding is not sentinel:
        self._encoding = encoding
        logging.debug(f"Changed encoding of FileStorage to {encoding}.")

    if cache is not sentinel:
        self._cache = cache  # type: ignore
        logging.debug(f"Changed cache of FileStorage to {cache}.")
Last update: August 5, 2022
Created: August 5, 2022