Skip to content

ReaderBase

Bases: ABC

Source code in magic_storage/base/_reader.py 
 15
 16
 17
 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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
class ReaderBase(ABC):
    def is_available(self, __uid: str, /) -> bool:
        """Check if object with specified identifier and store type is present
        in cache.

        Parameters
        ----------
        uid : str
            object unique identifier.

        Returns
        -------
        bool
            True when object is present, False otherwise.

        Examples
        --------
        ```
        >>> ReaderExampleImpl().is_available("example1")
        True
        >>> ReaderExampleImpl().is_available("not available")
        False
        >>>
        ```
        """
        __uid = make_uid(__uid)
        assert isinstance(__uid, str), __uid

        status = self._is_available(__uid)
        logging.debug(f"Availability status of {__uid} is {status}.")

        return status

    @abstractmethod
    def _is_available(self, __uid: str, /) -> bool:
        ...

    def load_as(  # noqa: FNE004
        self,
        store_type: StoreType,
        uid: str,
        **load_kw: Any,
    ) -> Any:
        """Load object from cache in format selected by parameter store_as.

        Parameters
        ----------
        store_type : StoreType, optional
            store type from enum.
        uid : str
            object unique identifier.

        Returns
        -------
        str
            Loaded object.

        Examples
        --------
        ```
        >>> ReaderExampleImpl().load_as(StoreType.TEXT, uid="example1")
        '{"foo": 32}'
        >>> ReaderExampleImpl().load_as(StoreType.PICKLE, uid="example2")
        {'foo': 32}
        >>>
        ```
        """
        return self._load_as(store_type, uid=uid, **load_kw)

    def _load_as(  # noqa: FNE004
        self,
        store_type: StoreType,
        /,
        *,
        uid: str,
        **load_kw: Any,
    ) -> Any:
        uid = make_uid(uid)
        assert isinstance(uid, str), uid
        logging.debug(f"Loading '{uid}' as {store_type}.")

        # We can't check if retval is not None as anything can be stored, including None
        retval = self._LOAD_MAP[store_type](self, uid, **load_kw)

        logging.debug(f"Successfully loaded {uid} as {store_type}")
        return retval

    def _load_text(self, uid: str) -> str:  # noqa: FNE004
        value = self._read_text(uid)
        assert isinstance(value, str)

        return value

    @abstractmethod
    def _read_text(self, __uid: str, /) -> str:
        ...

    def _load_json(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
        raw_value = self._read_text(uid)
        assert isinstance(raw_value, str)

        value = json.loads(raw_value, **load_kw)
        return value

    def _load_bytes(self, uid: str) -> bytes:  # noqa: FNE004
        value = self._read_bytes(uid)
        assert isinstance(value, bytes), value

        return value

    @abstractmethod
    def _read_bytes(self, __uid: str, /) -> bytes:
        ...

    def _load_pickle(  # noqa: FNE004
        self, uid: str, **pickle_load_kw: Any
    ) -> Any:
        source = self._read_bytes(uid)
        assert isinstance(source, bytes)

        source = decompress(source)
        assert isinstance(source, bytes)

        ob = pickle.loads(source, **pickle_load_kw)
        return ob

    def load_text(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
        """Load object with specified identifier as text.

        This is only suitable for str and str-like objects.

        Parameters
        ----------
        uid : str
            object unique identifier. Only Alphanumeric characters are allowed,
            other are replaced with '_'.

        Returns
        -------
        str
            Loaded object.
        """
        return self._load_as(StoreType.TEXT, uid=uid, **load_kw)

    def load_bytes(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
        """Load object with specified identifier as binary.

        This is only suitable for bytes and bytes-like objects.

        Parameters
        ----------
        uid : str
            object unique identifier.

        Returns
        -------
        bytes
            Loaded object.
        """
        return self._load_as(StoreType.BINARY, uid=uid, **load_kw)

    def load_json(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
        """Load object with specified identifier as json.

        This can be used with any object which can be decoded with json.loads()

        Parameters
        ----------
        uid : str
            object unique identifier. Only Alphanumeric characters are allowed,
            other are replaced with '_'.
        **load_kw: Any
            keyword argument passed to json.loads().

        Returns
        -------
        Any
            Loaded object.
        """
        return self._load_as(StoreType.JSON, uid=uid, **load_kw)

    def load_pickle(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
        """Load object with specified identifier as pickle.

        This can be used with any object which can be decoded with pickle.loads()

        Parameters
        ----------
        uid : str
            object unique identifier.
        **load_kw: Any
            keyword argument passed to pickle.loads().

        Returns
        -------
        Any
            Loaded object.
        """
        return self._load_as(StoreType.PICKLE, uid=uid, **load_kw)

    _LOAD_MAP: dict[StoreType, Callable[[ReaderBase, str], Any]] = {
        StoreType.TEXT: _load_text,
        StoreType.BINARY: _load_bytes,
        StoreType.JSON: _load_json,
        StoreType.PICKLE: _load_pickle,
    }

is_available(__uid) #

Check if object with specified identifier and store type is present in cache.

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required

Returns:

Type Description
bool

True when object is present, False otherwise.

Examples:

1
2
3
4
5
>>> ReaderExampleImpl().is_available("example1")
True
>>> ReaderExampleImpl().is_available("not available")
False
>>>
Source code in magic_storage/base/_reader.py 
16
17
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
def is_available(self, __uid: str, /) -> bool:
    """Check if object with specified identifier and store type is present
    in cache.

    Parameters
    ----------
    uid : str
        object unique identifier.

    Returns
    -------
    bool
        True when object is present, False otherwise.

    Examples
    --------
    ```
    >>> ReaderExampleImpl().is_available("example1")
    True
    >>> ReaderExampleImpl().is_available("not available")
    False
    >>>
    ```
    """
    __uid = make_uid(__uid)
    assert isinstance(__uid, str), __uid

    status = self._is_available(__uid)
    logging.debug(f"Availability status of {__uid} is {status}.")

    return status

load_as(store_type, uid, **load_kw) #

Load object from cache in format selected by parameter store_as.

Parameters:

Name Type Description Default
store_type StoreType, optional

store type from enum.

 required
uid str

object unique identifier.

 required

Returns:

Type Description
str

Loaded object.

Examples:

1
2
3
4
5
>>> ReaderExampleImpl().load_as(StoreType.TEXT, uid="example1")
'{"foo": 32}'
>>> ReaderExampleImpl().load_as(StoreType.PICKLE, uid="example2")
{'foo': 32}
>>>
Source code in magic_storage/base/_reader.py 
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
def load_as(  # noqa: FNE004
    self,
    store_type: StoreType,
    uid: str,
    **load_kw: Any,
) -> Any:
    """Load object from cache in format selected by parameter store_as.

    Parameters
    ----------
    store_type : StoreType, optional
        store type from enum.
    uid : str
        object unique identifier.

    Returns
    -------
    str
        Loaded object.

    Examples
    --------
    ```
    >>> ReaderExampleImpl().load_as(StoreType.TEXT, uid="example1")
    '{"foo": 32}'
    >>> ReaderExampleImpl().load_as(StoreType.PICKLE, uid="example2")
    {'foo': 32}
    >>>
    ```
    """
    return self._load_as(store_type, uid=uid, **load_kw)

load_bytes(uid, **load_kw) #

Load object with specified identifier as binary.

This is only suitable for bytes and bytes-like objects.

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required

Returns:

Type Description
bytes

Loaded object.
Source code in magic_storage/base/_reader.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def load_bytes(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
    """Load object with specified identifier as binary.

    This is only suitable for bytes and bytes-like objects.

    Parameters
    ----------
    uid : str
        object unique identifier.

    Returns
    -------
    bytes
        Loaded object.
    """
    return self._load_as(StoreType.BINARY, uid=uid, **load_kw)

load_json(uid, **load_kw) #

Load object with specified identifier as json.

This can be used with any object which can be decoded with json.loads()

Parameters:

Name Type Description Default
uid str

object unique identifier. Only Alphanumeric characters are allowed, other are replaced with '_'.

 required
**load_kw Any

keyword argument passed to json.loads().

 {}

Returns:

Type Description
Any

Loaded object.
Source code in magic_storage/base/_reader.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
def load_json(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
    """Load object with specified identifier as json.

    This can be used with any object which can be decoded with json.loads()

    Parameters
    ----------
    uid : str
        object unique identifier. Only Alphanumeric characters are allowed,
        other are replaced with '_'.
    **load_kw: Any
        keyword argument passed to json.loads().

    Returns
    -------
    Any
        Loaded object.
    """
    return self._load_as(StoreType.JSON, uid=uid, **load_kw)

load_pickle(uid, **load_kw) #

Load object with specified identifier as pickle.

This can be used with any object which can be decoded with pickle.loads()

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required
**load_kw Any

keyword argument passed to pickle.loads().

 {}

Returns:

Type Description
Any

Loaded object.
Source code in magic_storage/base/_reader.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def load_pickle(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
    """Load object with specified identifier as pickle.

    This can be used with any object which can be decoded with pickle.loads()

    Parameters
    ----------
    uid : str
        object unique identifier.
    **load_kw: Any
        keyword argument passed to pickle.loads().

    Returns
    -------
    Any
        Loaded object.
    """
    return self._load_as(StoreType.PICKLE, uid=uid, **load_kw)

load_text(uid, **load_kw) #

Load object with specified identifier as text.

This is only suitable for str and str-like objects.

Parameters:

Name Type Description Default
uid str

object unique identifier. Only Alphanumeric characters are allowed, other are replaced with '_'.

 required

Returns:

Type Description
str

Loaded object.
Source code in magic_storage/base/_reader.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def load_text(self, uid: str, **load_kw: Any) -> Any:  # noqa: FNE004
    """Load object with specified identifier as text.

    This is only suitable for str and str-like objects.

    Parameters
    ----------
    uid : str
        object unique identifier. Only Alphanumeric characters are allowed,
        other are replaced with '_'.

    Returns
    -------
    str
        Loaded object.
    """
    return self._load_as(StoreType.TEXT, uid=uid, **load_kw)
Last update: August 5, 2022
Created: August 5, 2022