Skip to content

WriterBase

Bases: ABC

Source code in magic_storage/base/_writer.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
class WriterBase(ABC):
    def store_as(
        self,
        store_type: StoreType,
        /,
        uid: str,
        item: Any,
        **dump_kw: Any,
    ) -> str:
        """Dump object to cache in format selected by parameter store_as.

        Parameters
        ----------
        store_type : StoreType
            store type from enum
        uid : str
            object unique identifier.
        item : Any
            item to store, constraints depend on storage type.

        Returns
        -------
        str
            Identifier after cleanup and tagging (real used identifier).
        """
        return self._store_as(store_type, uid=uid, item=item, **dump_kw)

    def _store_as(
        self,
        store_type: StoreType,
        /,
        *,
        uid: str,
        item: Any,
        **dump_kw: Any,
    ) -> str:
        uid = make_uid(uid)
        assert isinstance(uid, str), uid
        logging.debug(f"Dumping '{uid}' as {store_type}.")

        retval = self._STORE_MAP[store_type](self, uid, item, **dump_kw)
        assert retval is None, retval

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

    def _store_text(self, uid: str, item: Any, **str_kw: Any) -> None:
        raw_value = str(item, **str_kw)
        assert isinstance(raw_value, str), raw_value

        retval = self._write_text(uid, raw_value)
        assert retval is None, retval

        return retval

    @abstractmethod
    def _write_text(self, __uid: str, __item: str, /) -> None:
        ...

    def _store_json(self, uid: str, item: Any, **json_dumps_kw: Any) -> None:
        try:
            raw_value = json.dumps(item, **json_dumps_kw)
        except TypeError:
            if hasattr(item, "json") and callable(item.json):
                raw_value = item.json()
            else:
                raise
        assert isinstance(raw_value, str), raw_value

        retval = self._write_text(uid, raw_value)
        assert retval is None, retval

        return retval

    def _store_bytes(self, uid: str, item: Any, **bytes_kw: Any) -> None:
        raw_value = bytes(item, **bytes_kw)
        assert isinstance(raw_value, bytes), raw_value

        retval = self._write_bytes(uid, raw_value)
        assert retval is None, retval

        logging.debug(f"Successfully dumped {uid} as BYTES")
        return retval

    @abstractmethod
    def _write_bytes(self, __uid: str, __item: bytes, /) -> None:
        ...

    def _store_pickle(
        self, uid: str, item: Any, **pickle_dump_kw: Any
    ) -> None:
        raw_value = pickle.dumps(item, **pickle_dump_kw)
        assert isinstance(raw_value, bytes)

        raw_value = compress(raw_value)
        assert isinstance(raw_value, bytes), raw_value

        retval = self._write_bytes(uid, raw_value)
        assert retval is None, retval

        logging.debug(f"Successfully dumped {uid} as PICKLE")
        return retval

    def store_str(self, uid: str, item: str, **str_kw: Any) -> str:
        """Dump object to cache in form of text.

        Parameters
        ----------
        uid : str
            object unique identifier.
        item : str
            item to store, str or str-like objects.

        Returns
        -------
        str
            Identifier after cleanup and tagging (real used identifier).
        """
        return self._store_as(StoreType.TEXT, uid=uid, item=item, **str_kw)

    def store_bytes(self, uid: str, item: bytes, **bytes_kw: Any) -> str:
        """Dump object to cache in form of binary.

        Parameters
        ----------
        uid : str
            object unique identifier.
        item : str
            item to store, bytes or bytes-like objects.

        Returns
        -------
        str
            Identifier after cleanup and tagging (real used identifier).
        """
        return self._store_as(StoreType.BINARY, uid=uid, item=item, **bytes_kw)

    def store_json(self, uid: str, item: Any, **json_dumps_kw: Any) -> str:
        """Dump object to cache in form of json encoded text.

        Parameters
        ----------
        uid : str
            object unique identifier.
        item : str
            item to store, any object which can be encoded with json.dumps().

        Returns
        -------
        str
            Identifier after cleanup and tagging (real used identifier).
        """
        return self._store_as(
            StoreType.JSON, uid=uid, item=item, **json_dumps_kw
        )

    def store_pickle(self, uid: str, item: Any, **pickle_dump_kw: Any) -> str:
        """Dump object to cache in form of pickled binary.

        Because pickle is a binary format, it is always compressed with lzma algorithm.

        Parameters
        ----------
        identifier : str
            object unique identifier.
        item : str
            item to store, any object which can be encoded with pickle.dumps().


        Returns
        -------
        str
            Identifier after cleanup and tagging (real used identifier).
        """
        return self._store_as(
            StoreType.PICKLE, uid=uid, item=item, **pickle_dump_kw
        )

    _STORE_MAP: dict[StoreType, Callable[[WriterBase, str, Any], None]] = {
        StoreType.TEXT: _store_text,
        StoreType.BINARY: _store_bytes,
        StoreType.JSON: _store_json,
        StoreType.PICKLE: _store_pickle,
    }

store_as(store_type, /, uid, item, **dump_kw) #

Dump object to cache in format selected by parameter store_as.

Parameters:

Name Type Description Default
store_type StoreType

store type from enum

 required
uid str

object unique identifier.

 required
item Any

item to store, constraints depend on storage type.

 required

Returns:

Type Description
str

Identifier after cleanup and tagging (real used identifier).
Source code in magic_storage/base/_writer.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
def store_as(
    self,
    store_type: StoreType,
    /,
    uid: str,
    item: Any,
    **dump_kw: Any,
) -> str:
    """Dump object to cache in format selected by parameter store_as.

    Parameters
    ----------
    store_type : StoreType
        store type from enum
    uid : str
        object unique identifier.
    item : Any
        item to store, constraints depend on storage type.

    Returns
    -------
    str
        Identifier after cleanup and tagging (real used identifier).
    """
    return self._store_as(store_type, uid=uid, item=item, **dump_kw)

store_bytes(uid, item, **bytes_kw) #

Dump object to cache in form of binary.

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required
item str

item to store, bytes or bytes-like objects.

 required

Returns:

Type Description
str

Identifier after cleanup and tagging (real used identifier).
Source code in magic_storage/base/_writer.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def store_bytes(self, uid: str, item: bytes, **bytes_kw: Any) -> str:
    """Dump object to cache in form of binary.

    Parameters
    ----------
    uid : str
        object unique identifier.
    item : str
        item to store, bytes or bytes-like objects.

    Returns
    -------
    str
        Identifier after cleanup and tagging (real used identifier).
    """
    return self._store_as(StoreType.BINARY, uid=uid, item=item, **bytes_kw)

store_json(uid, item, **json_dumps_kw) #

Dump object to cache in form of json encoded text.

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required
item str

item to store, any object which can be encoded with json.dumps().

 required

Returns:

Type Description
str

Identifier after cleanup and tagging (real used identifier).
Source code in magic_storage/base/_writer.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def store_json(self, uid: str, item: Any, **json_dumps_kw: Any) -> str:
    """Dump object to cache in form of json encoded text.

    Parameters
    ----------
    uid : str
        object unique identifier.
    item : str
        item to store, any object which can be encoded with json.dumps().

    Returns
    -------
    str
        Identifier after cleanup and tagging (real used identifier).
    """
    return self._store_as(
        StoreType.JSON, uid=uid, item=item, **json_dumps_kw
    )

store_pickle(uid, item, **pickle_dump_kw) #

Dump object to cache in form of pickled binary.

Because pickle is a binary format, it is always compressed with lzma algorithm.

Parameters:

Name Type Description Default
identifier str

object unique identifier.

 required
item str

item to store, any object which can be encoded with pickle.dumps().

 required

Returns:

Type Description
str

Identifier after cleanup and tagging (real used identifier).
Source code in magic_storage/base/_writer.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def store_pickle(self, uid: str, item: Any, **pickle_dump_kw: Any) -> str:
    """Dump object to cache in form of pickled binary.

    Because pickle is a binary format, it is always compressed with lzma algorithm.

    Parameters
    ----------
    identifier : str
        object unique identifier.
    item : str
        item to store, any object which can be encoded with pickle.dumps().


    Returns
    -------
    str
        Identifier after cleanup and tagging (real used identifier).
    """
    return self._store_as(
        StoreType.PICKLE, uid=uid, item=item, **pickle_dump_kw
    )

store_str(uid, item, **str_kw) #

Dump object to cache in form of text.

Parameters:

Name Type Description Default
uid str

object unique identifier.

 required
item str

item to store, str or str-like objects.

 required

Returns:

Type Description
str

Identifier after cleanup and tagging (real used identifier).
Source code in magic_storage/base/_writer.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def store_str(self, uid: str, item: str, **str_kw: Any) -> str:
    """Dump object to cache in form of text.

    Parameters
    ----------
    uid : str
        object unique identifier.
    item : str
        item to store, str or str-like objects.

    Returns
    -------
    str
        Identifier after cleanup and tagging (real used identifier).
    """
    return self._store_as(StoreType.TEXT, uid=uid, item=item, **str_kw)
Last update: August 5, 2022
Created: August 5, 2022