Skip to content

Storage

framework3.base.base_storage

__all__ = ['BaseStorage', 'BaseSingleton'] module-attribute

BaseSingleton

A base class for implementing the Singleton pattern.

This class ensures that only one instance of each derived class is created.

Key Features
  • Implements the Singleton design pattern
  • Allows derived classes to have only one instance
Usage

To create a Singleton class, inherit from BaseSingleton:

class MySingleton(BaseSingleton):
    def __init__(self):
        self.value = 0

    def increment(self):
        self.value += 1

# Usage
instance1 = MySingleton()
instance2 = MySingleton()
assert instance1 is instance2  # True

Attributes:

Name Type Description
_instances Dict[Type[BaseSingleton], Any]

A class-level dictionary to store instances.

Methods:

Name Description
__new__

Type[BaseSingleton], args: Any, *kwargs: Any) -> BaseStorage: Creates a new instance or returns the existing one.
Note

This class should be used as a base class for any class that needs to implement the Singleton pattern.

Source code in framework3/base/base_storage.py 
11
12
13
14
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
class BaseSingleton:
    """
    A base class for implementing the Singleton pattern.

    This class ensures that only one instance of each derived class is created.

    Key Features:
        - Implements the Singleton design pattern
        - Allows derived classes to have only one instance

    Usage:
        To create a Singleton class, inherit from BaseSingleton:

        ```python
        class MySingleton(BaseSingleton):
            def __init__(self):
                self.value = 0

            def increment(self):
                self.value += 1

        # Usage
        instance1 = MySingleton()
        instance2 = MySingleton()
        assert instance1 is instance2  # True
        ```

    Attributes:
        _instances (Dict[Type[BaseSingleton], Any]): A class-level dictionary to store instances.

    Methods:
        __new__(cls: Type[BaseSingleton], *args: Any, **kwargs: Any) -> BaseStorage:
            Creates a new instance or returns the existing one.

    Note:
        This class should be used as a base class for any class that needs to implement
        the Singleton pattern.
    """

    _instances: Dict[Type[BaseSingleton], Any] = {}

    def __new__(cls: Type[BaseSingleton], *args: Any, **kwargs: Any) -> BaseStorage:
        """
        Create a new instance of the class if it doesn't exist, otherwise return the existing instance.

        This method implements the core logic of the Singleton pattern.

        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.

        Returns:
            BaseStorage: The single instance of the class.

        Note:
            This method is called before __init__ when creating a new instance of the class.
        """
        if cls not in cls._instances:
            cls._instances[cls] = super().__new__(cls)  # type: ignore
        return cls._instances[cls]
__new__(*args, **kwargs)

Create a new instance of the class if it doesn't exist, otherwise return the existing instance.

This method implements the core logic of the Singleton pattern.

Parameters:

Name Type Description Default
*args Any

Variable length argument list.

 ()
**kwargs Any

Arbitrary keyword arguments.

 {}

Returns:

Name Type Description
BaseStorage BaseStorage

The single instance of the class.
Note

This method is called before init when creating a new instance of the class.

Source code in framework3/base/base_storage.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def __new__(cls: Type[BaseSingleton], *args: Any, **kwargs: Any) -> BaseStorage:
    """
    Create a new instance of the class if it doesn't exist, otherwise return the existing instance.

    This method implements the core logic of the Singleton pattern.

    Args:
        *args: Variable length argument list.
        **kwargs: Arbitrary keyword arguments.

    Returns:
        BaseStorage: The single instance of the class.

    Note:
        This method is called before __init__ when creating a new instance of the class.
    """
    if cls not in cls._instances:
        cls._instances[cls] = super().__new__(cls)  # type: ignore
    return cls._instances[cls]

BaseStorage

Bases: BasePlugin, BaseSingleton

An abstract base class for storage operations.

This class defines the interface for storage-related operations and inherits from BasePlugin for plugin functionality and BaseSingleton for single instance behavior.

Key Features
  • Abstract methods for common storage operations
  • Singleton behavior ensures only one instance per storage type
  • Inherits plugin functionality from BasePlugin
Usage

To create a new storage type, inherit from BaseStorage and implement all abstract methods:

class MyCustomStorage(BaseStorage):
    def __init__(self, root_path: str):
        self.root_path = root_path

    def get_root_path(self) -> str:
        return self.root_path

    def upload_file(self, file, file_name: str, context: str, direct_stream: bool = False) -> str | None:
        # Implement file upload logic
        ...

    # Implement other abstract methods
    ...

# Usage
storage = MyCustomStorage("/path/to/storage")
storage.upload_file(file_object, "example.txt", "documents")

Methods:

Name Description
get_root_path

Abstract method to get the root path of the storage.
upload_file

object, file_name: str, context: str, direct_stream: bool = False) -> str | None: Abstract method to upload a file to the storage.
download_file

str, context: str) -> Any: Abstract method to download a file from the storage.
list_stored_files

str) -> List[Any]: Abstract method to list all files stored in a specific context.
get_file_by_hashcode

str, context: str) -> Any: Abstract method to retrieve a file by its hashcode.
check_if_exists

str, context: str) -> bool: Abstract method to check if a file exists in the storage.
delete_file

str, context: str): Abstract method to delete a file from the storage.
Note

This is an abstract base class. Concrete implementations should override all abstract methods to provide specific storage functionality.

Source code in framework3/base/base_storage.py 
 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
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
class BaseStorage(BasePlugin, BaseSingleton):
    """
    An abstract base class for storage operations.

    This class defines the interface for storage-related operations and inherits
    from BasePlugin for plugin functionality and BaseSingleton for single instance behavior.

    Key Features:
        - Abstract methods for common storage operations
        - Singleton behavior ensures only one instance per storage type
        - Inherits plugin functionality from BasePlugin

    Usage:
        To create a new storage type, inherit from BaseStorage and implement all abstract methods:

        ```python
        class MyCustomStorage(BaseStorage):
            def __init__(self, root_path: str):
                self.root_path = root_path

            def get_root_path(self) -> str:
                return self.root_path

            def upload_file(self, file, file_name: str, context: str, direct_stream: bool = False) -> str | None:
                # Implement file upload logic
                ...

            # Implement other abstract methods
            ...

        # Usage
        storage = MyCustomStorage("/path/to/storage")
        storage.upload_file(file_object, "example.txt", "documents")
        ```

    Methods:
        get_root_path() -> str:
            Abstract method to get the root path of the storage.
        upload_file(file: object, file_name: str, context: str, direct_stream: bool = False) -> str | None:
            Abstract method to upload a file to the storage.
        download_file(hashcode: str, context: str) -> Any:
            Abstract method to download a file from the storage.
        list_stored_files(context: str) -> List[Any]:
            Abstract method to list all files stored in a specific context.
        get_file_by_hashcode(hashcode: str, context: str) -> Any:
            Abstract method to retrieve a file by its hashcode.
        check_if_exists(hashcode: str, context: str) -> bool:
            Abstract method to check if a file exists in the storage.
        delete_file(hashcode: str, context: str):
            Abstract method to delete a file from the storage.

    Note:
        This is an abstract base class. Concrete implementations should override
        all abstract methods to provide specific storage functionality.
    """

    @abstractmethod
    def get_root_path(self) -> str:
        """
        Get the root path of the storage.

        This method should be implemented to return the base directory or path
        where the storage system keeps its files.

        Returns:
            str: The root path of the storage.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def get_root_path(self) -> str:
                return "/var/data/storage"
            ```
        """
        ...

    @abstractmethod
    def upload_file(
        self, file: object, file_name: str, context: str, direct_stream: bool = False
    ) -> str | None:
        """
        Upload a file to the storage.

        This method should be implemented to handle file uploads to the storage system.

        Args:
            file (object): The file object to upload.
            file_name (str): The name of the file.
            context (str): The context or directory for the file.
            direct_stream (bool, optional): Whether to use direct streaming. Defaults to False.

        Returns:
            str | None: The identifier of the uploaded file, or None if upload failed.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def upload_file(self, file: object, file_name: str, context: str, direct_stream: bool = False) -> str | None:
                path = os.path.join(self.get_root_path(), context, file_name)
                with open(path, 'wb') as f:
                    f.write(file.read())
                return file_name
            ```
        """
        ...

    @abstractmethod
    def download_file(self, hashcode: str, context: str) -> Any:
        """
        Download a file from the storage.

        This method should be implemented to retrieve files from the storage system.

        Args:
            hashcode (str): The identifier of the file to download.
            context (str): The context or directory of the file.

        Returns:
            Any: The downloaded file object.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def download_file(self, hashcode: str, context: str) -> Any:
                path = os.path.join(self.get_root_path(), context, hashcode)
                with open(path, 'rb') as f:
                    return f.read()
            ```
        """
        ...

    @abstractmethod
    def list_stored_files(self, context: str) -> List[Any]:
        """
        List all files stored in a specific context.

        This method should be implemented to return a list of files in a given context.

        Args:
            context (str): The context or directory to list files from.

        Returns:
            List[Any]: A list of file objects or file information.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def list_stored_files(self, context: str) -> List[Any]:
                path = os.path.join(self.get_root_path(), context)
                return os.listdir(path)
            ```
        """
        ...

    @abstractmethod
    def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
        """
        Retrieve a file by its hashcode.

        This method should be implemented to fetch a specific file using its identifier.

        Args:
            hashcode (str): The identifier of the file.
            context (str): The context or directory of the file.

        Returns:
            Any: The file object or file information.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
                return self.download_file(hashcode, context)
            ```
        """
        ...

    @abstractmethod
    def check_if_exists(self, hashcode: str, context: str) -> bool:
        """
        Check if a file exists in the storage.

        This method should be implemented to verify the existence of a file.

        Args:
            hashcode (str): The identifier of the file.
            context (str): The context or directory of the file.

        Returns:
            bool: True if the file exists, False otherwise.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def check_if_exists(self, hashcode: str, context: str) -> bool:
                path = os.path.join(self.get_root_path(), context, hashcode)
                return os.path.exists(path)
            ```
        """
        ...

    @abstractmethod
    def delete_file(self, hashcode: str, context: str):
        """
        Delete a file from the storage.

        This method should be implemented to remove a file from the storage system.

        Args:
            hashcode (str): The identifier of the file to delete.
            context (str): The context or directory of the file.

        Raises:
            NotImplementedError: If the subclass does not implement this method.

        Example:
            ```python
            def delete_file(self, hashcode: str, context: str):
                path = os.path.join(self.get_root_path(), context, hashcode)
                if os.path.exists(path):
                    os.remove(path)
                else:
                    raise FileNotFoundError(f"File {hashcode} not found in {context}")
            ```
        """
        ...
check_if_exists(hashcode, context) abstractmethod

Check if a file exists in the storage.

This method should be implemented to verify the existence of a file.

Parameters:

Name Type Description Default
hashcode str

The identifier of the file.

 required
context str

The context or directory of the file.

 required

Returns:

Name Type Description
bool bool

True if the file exists, False otherwise.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def check_if_exists(self, hashcode: str, context: str) -> bool:
    path = os.path.join(self.get_root_path(), context, hashcode)
    return os.path.exists(path)
Source code in framework3/base/base_storage.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
@abstractmethod
def check_if_exists(self, hashcode: str, context: str) -> bool:
    """
    Check if a file exists in the storage.

    This method should be implemented to verify the existence of a file.

    Args:
        hashcode (str): The identifier of the file.
        context (str): The context or directory of the file.

    Returns:
        bool: True if the file exists, False otherwise.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def check_if_exists(self, hashcode: str, context: str) -> bool:
            path = os.path.join(self.get_root_path(), context, hashcode)
            return os.path.exists(path)
        ```
    """
    ...
delete_file(hashcode, context) abstractmethod

Delete a file from the storage.

This method should be implemented to remove a file from the storage system.

Parameters:

Name Type Description Default
hashcode str

The identifier of the file to delete.

 required
context str

The context or directory of the file.

 required

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def delete_file(self, hashcode: str, context: str):
    path = os.path.join(self.get_root_path(), context, hashcode)
    if os.path.exists(path):
        os.remove(path)
    else:
        raise FileNotFoundError(f"File {hashcode} not found in {context}")
Source code in framework3/base/base_storage.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
@abstractmethod
def delete_file(self, hashcode: str, context: str):
    """
    Delete a file from the storage.

    This method should be implemented to remove a file from the storage system.

    Args:
        hashcode (str): The identifier of the file to delete.
        context (str): The context or directory of the file.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def delete_file(self, hashcode: str, context: str):
            path = os.path.join(self.get_root_path(), context, hashcode)
            if os.path.exists(path):
                os.remove(path)
            else:
                raise FileNotFoundError(f"File {hashcode} not found in {context}")
        ```
    """
    ...
download_file(hashcode, context) abstractmethod

Download a file from the storage.

This method should be implemented to retrieve files from the storage system.

Parameters:

Name Type Description Default
hashcode str

The identifier of the file to download.

 required
context str

The context or directory of the file.

 required

Returns:

Name Type Description
Any Any

The downloaded file object.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def download_file(self, hashcode: str, context: str) -> Any:
    path = os.path.join(self.get_root_path(), context, hashcode)
    with open(path, 'rb') as f:
        return f.read()
Source code in framework3/base/base_storage.py 
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
@abstractmethod
def download_file(self, hashcode: str, context: str) -> Any:
    """
    Download a file from the storage.

    This method should be implemented to retrieve files from the storage system.

    Args:
        hashcode (str): The identifier of the file to download.
        context (str): The context or directory of the file.

    Returns:
        Any: The downloaded file object.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def download_file(self, hashcode: str, context: str) -> Any:
            path = os.path.join(self.get_root_path(), context, hashcode)
            with open(path, 'rb') as f:
                return f.read()
        ```
    """
    ...
get_file_by_hashcode(hashcode, context) abstractmethod

Retrieve a file by its hashcode.

This method should be implemented to fetch a specific file using its identifier.

Parameters:

Name Type Description Default
hashcode str

The identifier of the file.

 required
context str

The context or directory of the file.

 required

Returns:

Name Type Description
Any Any

The file object or file information.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
    return self.download_file(hashcode, context)
Source code in framework3/base/base_storage.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
@abstractmethod
def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
    """
    Retrieve a file by its hashcode.

    This method should be implemented to fetch a specific file using its identifier.

    Args:
        hashcode (str): The identifier of the file.
        context (str): The context or directory of the file.

    Returns:
        Any: The file object or file information.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
            return self.download_file(hashcode, context)
        ```
    """
    ...
get_root_path() abstractmethod

Get the root path of the storage.

This method should be implemented to return the base directory or path where the storage system keeps its files.

Returns:

Name Type Description
str str

The root path of the storage.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def get_root_path(self) -> str:
    return "/var/data/storage"
Source code in framework3/base/base_storage.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@abstractmethod
def get_root_path(self) -> str:
    """
    Get the root path of the storage.

    This method should be implemented to return the base directory or path
    where the storage system keeps its files.

    Returns:
        str: The root path of the storage.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def get_root_path(self) -> str:
            return "/var/data/storage"
        ```
    """
    ...
list_stored_files(context) abstractmethod

List all files stored in a specific context.

This method should be implemented to return a list of files in a given context.

Parameters:

Name Type Description Default
context str

The context or directory to list files from.

 required

Returns:

Type Description
List[Any]

List[Any]: A list of file objects or file information.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def list_stored_files(self, context: str) -> List[Any]:
    path = os.path.join(self.get_root_path(), context)
    return os.listdir(path)
Source code in framework3/base/base_storage.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
@abstractmethod
def list_stored_files(self, context: str) -> List[Any]:
    """
    List all files stored in a specific context.

    This method should be implemented to return a list of files in a given context.

    Args:
        context (str): The context or directory to list files from.

    Returns:
        List[Any]: A list of file objects or file information.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def list_stored_files(self, context: str) -> List[Any]:
            path = os.path.join(self.get_root_path(), context)
            return os.listdir(path)
        ```
    """
    ...
upload_file(file, file_name, context, direct_stream=False) abstractmethod

Upload a file to the storage.

This method should be implemented to handle file uploads to the storage system.

Parameters:

Name Type Description Default
file object

The file object to upload.

 required
file_name str

The name of the file.

 required
context str

The context or directory for the file.

 required
direct_stream bool

Whether to use direct streaming. Defaults to False.

 False

Returns:

Type Description
str | None

str | None: The identifier of the uploaded file, or None if upload failed.

Raises:

Type Description
NotImplementedError

If the subclass does not implement this method.
Example 
def upload_file(self, file: object, file_name: str, context: str, direct_stream: bool = False) -> str | None:
    path = os.path.join(self.get_root_path(), context, file_name)
    with open(path, 'wb') as f:
        f.write(file.read())
    return file_name
Source code in framework3/base/base_storage.py 
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
@abstractmethod
def upload_file(
    self, file: object, file_name: str, context: str, direct_stream: bool = False
) -> str | None:
    """
    Upload a file to the storage.

    This method should be implemented to handle file uploads to the storage system.

    Args:
        file (object): The file object to upload.
        file_name (str): The name of the file.
        context (str): The context or directory for the file.
        direct_stream (bool, optional): Whether to use direct streaming. Defaults to False.

    Returns:
        str | None: The identifier of the uploaded file, or None if upload failed.

    Raises:
        NotImplementedError: If the subclass does not implement this method.

    Example:
        ```python
        def upload_file(self, file: object, file_name: str, context: str, direct_stream: bool = False) -> str | None:
            path = os.path.join(self.get_root_path(), context, file_name)
            with open(path, 'wb') as f:
                f.write(file.read())
            return file_name
        ```
    """
    ...