Skip to content

Local

framework3.plugins.storage.local_storage.LocalStorage

Bases: BaseStorage

A local file system storage implementation for storing and retrieving files.

This class provides methods to interact with the local file system, allowing storage operations such as uploading, downloading, and deleting files.

Key Features
  • Simple interface for file operations
  • Support for creating nested directory structures
  • File existence checking
  • Listing stored files in a given context
Usage 
from framework3.plugins.storage import LocalStorage

# Initialize local storage
storage = LocalStorage(storage_path='my_cache')

# Upload a file
storage.upload_file("Hello, World!", "greeting.txt", "/tmp")

# Download and read a file
content = storage.download_file("greeting.txt", "/tmp")
print(content)  # Output: Hello, World!

# Check if a file exists
exists = storage.check_if_exists("greeting.txt", "/tmp")
print(exists)  # Output: True

# List files in a directory
files = storage.list_stored_files("/tmp")
print(files)  # Output: ['greeting.txt']

# Delete a file
storage.delete_file("greeting.txt", "/tmp")

Attributes:

Name Type Description
storage_path str

The base path for storage operations.
_base_path str

The full path to the storage directory.

Methods:

Name Description
get_root_path

Get the root path of the storage.
upload_file

str, context: str, direct_stream: bool = False) -> str | None: Upload a file to the specified context.
list_stored_files

str) -> List[str]: List all files in the specified context.
get_file_by_hashcode

str, context: str) -> Any: Get a file by its hashcode.
check_if_exists

str, context: str) -> bool: Check if a file exists in the specified context.
download_file

str, context: str) -> Any: Download and load a file from the specified context.
delete_file

str, context: str) -> None: Delete a file from the specified context.
Source code in framework3/plugins/storage/local_storage.py 
 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
 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
class LocalStorage(BaseStorage):
    """
    A local file system storage implementation for storing and retrieving files.

    This class provides methods to interact with the local file system, allowing
    storage operations such as uploading, downloading, and deleting files.

    Key Features:
        - Simple interface for file operations
        - Support for creating nested directory structures
        - File existence checking
        - Listing stored files in a given context

    Usage:
        ```python
        from framework3.plugins.storage import LocalStorage

        # Initialize local storage
        storage = LocalStorage(storage_path='my_cache')

        # Upload a file
        storage.upload_file("Hello, World!", "greeting.txt", "/tmp")

        # Download and read a file
        content = storage.download_file("greeting.txt", "/tmp")
        print(content)  # Output: Hello, World!

        # Check if a file exists
        exists = storage.check_if_exists("greeting.txt", "/tmp")
        print(exists)  # Output: True

        # List files in a directory
        files = storage.list_stored_files("/tmp")
        print(files)  # Output: ['greeting.txt']

        # Delete a file
        storage.delete_file("greeting.txt", "/tmp")
        ```

    Attributes:
        storage_path (str): The base path for storage operations.
        _base_path (str): The full path to the storage directory.

    Methods:
        get_root_path() -> str: Get the root path of the storage.
        upload_file(file, file_name: str, context: str, direct_stream: bool = False) -> str | None:
            Upload a file to the specified context.
        list_stored_files(context: str) -> List[str]: List all files in the specified context.
        get_file_by_hashcode(hashcode: str, context: str) -> Any: Get a file by its hashcode.
        check_if_exists(hashcode: str, context: str) -> bool: Check if a file exists in the specified context.
        download_file(hashcode: str, context: str) -> Any: Download and load a file from the specified context.
        delete_file(hashcode: str, context: str) -> None: Delete a file from the specified context.
    """

    def __init__(self, storage_path: str = "cache"):
        """
        Initialize the LocalStorage.

        Args:
            storage_path (str, optional): The base path for storage. Defaults to 'cache'.
        """
        super().__init__()
        self.storage_path = storage_path
        self._base_path = storage_path

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

        Returns:
            str: The full path to the storage directory.
        """
        return self._base_path

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

        Args:
            file (Any): The file content to be uploaded.
            file_name (str): The name of the file.
            context (str): The directory path where the file will be saved.
            direct_stream (bool, optional): Not used in this implementation. Defaults to False.

        Returns:
            str | None: The file name if successful, None otherwise.
        """
        try:
            Path(context).mkdir(parents=True, exist_ok=True)
            print(f"\t * Saving in local path: {context}/{file_name}")
            pickle.dump(file, open(f"{context}/{file_name}", "wb"))
            print("\t * Saved !")
            return file_name
        except Exception as ex:
            print(ex)
        return None

    def list_stored_files(self, context: str) -> List[str]:
        """
        List all files in the specified context.

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

        Returns:
            List[str]: A list of file names in the specified context.
        """
        return os.listdir(context)

    def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
        """
        Get a file by its hashcode (filename in this implementation).

        Args:
            hashcode (str): The hashcode (filename) of the file.
            context (str): The directory path where the file is located.

        Returns:
            Any: A file object if found.

        Raises:
            FileNotFoundError: If the file is not found in the specified context.
        """
        if hashcode in os.listdir(context):
            return open(f"{context}/{hashcode}", "rb")
        else:
            raise FileNotFoundError(f"Couldn't find file {hashcode} in path {context}")

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

        Args:
            hashcode (str): The hashcode (filename) of the file.
            context (str): The directory path where to check for the file.

        Returns:
            bool: True if the file exists, False otherwise.
        """
        try:
            for file_n in os.listdir(context):
                if file_n == hashcode:
                    return True
            return False
        except FileNotFoundError:
            return False

    def download_file(self, hashcode: str, context: str) -> Any:
        """
        Download and load a file from the specified context.

        Args:
            hashcode (str): The hashcode (filename) of the file to download.
            context (str): The directory path where the file is located.

        Returns:
            Any: The content of the file, unpickled if it was pickled.
        """
        stream = self.get_file_by_hashcode(hashcode, context)
        print(f"\t * Downloading: {stream}")
        loaded = pickle.load(stream)
        return pickle.loads(loaded) if isinstance(loaded, bytes) else loaded

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

        Args:
            hashcode (str): The hashcode (filename) of the file to delete.
            context (str): The directory path where the file is located.

        Raises:
            FileExistsError: If the file does not exist in the specified context.
        """
        if os.path.exists(f"{context}/{hashcode}"):
            os.remove(f"{context}/{hashcode}")
        else:
            raise FileExistsError("No existe en la carpeta")

storage_path = storage_path instance-attribute

__init__(storage_path='cache')

Initialize the LocalStorage.

Parameters:

Name Type Description Default
storage_path str

The base path for storage. Defaults to 'cache'.

 'cache'
Source code in framework3/plugins/storage/local_storage.py 
66
67
68
69
70
71
72
73
74
75
def __init__(self, storage_path: str = "cache"):
    """
    Initialize the LocalStorage.

    Args:
        storage_path (str, optional): The base path for storage. Defaults to 'cache'.
    """
    super().__init__()
    self.storage_path = storage_path
    self._base_path = storage_path

check_if_exists(hashcode, context)

Check if a file exists in the specified context.

Parameters:

Name Type Description Default
hashcode str

The hashcode (filename) of the file.

 required
context str

The directory path where to check for the file.

 required

Returns:

Name Type Description
bool bool

True if the file exists, False otherwise.
Source code in framework3/plugins/storage/local_storage.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def check_if_exists(self, hashcode: str, context: str) -> bool:
    """
    Check if a file exists in the specified context.

    Args:
        hashcode (str): The hashcode (filename) of the file.
        context (str): The directory path where to check for the file.

    Returns:
        bool: True if the file exists, False otherwise.
    """
    try:
        for file_n in os.listdir(context):
            if file_n == hashcode:
                return True
        return False
    except FileNotFoundError:
        return False

delete_file(hashcode, context)

Delete a file from the specified context.

Parameters:

Name Type Description Default
hashcode str

The hashcode (filename) of the file to delete.

 required
context str

The directory path where the file is located.

 required

Raises:

Type Description
FileExistsError

If the file does not exist in the specified context.
Source code in framework3/plugins/storage/local_storage.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def delete_file(self, hashcode: str, context: str):
    """
    Delete a file from the specified context.

    Args:
        hashcode (str): The hashcode (filename) of the file to delete.
        context (str): The directory path where the file is located.

    Raises:
        FileExistsError: If the file does not exist in the specified context.
    """
    if os.path.exists(f"{context}/{hashcode}"):
        os.remove(f"{context}/{hashcode}")
    else:
        raise FileExistsError("No existe en la carpeta")

download_file(hashcode, context)

Download and load a file from the specified context.

Parameters:

Name Type Description Default
hashcode str

The hashcode (filename) of the file to download.

 required
context str

The directory path where the file is located.

 required

Returns:

Name Type Description
Any Any

The content of the file, unpickled if it was pickled.
Source code in framework3/plugins/storage/local_storage.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def download_file(self, hashcode: str, context: str) -> Any:
    """
    Download and load a file from the specified context.

    Args:
        hashcode (str): The hashcode (filename) of the file to download.
        context (str): The directory path where the file is located.

    Returns:
        Any: The content of the file, unpickled if it was pickled.
    """
    stream = self.get_file_by_hashcode(hashcode, context)
    print(f"\t * Downloading: {stream}")
    loaded = pickle.load(stream)
    return pickle.loads(loaded) if isinstance(loaded, bytes) else loaded

get_file_by_hashcode(hashcode, context)

Get a file by its hashcode (filename in this implementation).

Parameters:

Name Type Description Default
hashcode str

The hashcode (filename) of the file.

 required
context str

The directory path where the file is located.

 required

Returns:

Name Type Description
Any Any

A file object if found.

Raises:

Type Description
FileNotFoundError

If the file is not found in the specified context.
Source code in framework3/plugins/storage/local_storage.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_file_by_hashcode(self, hashcode: str, context: str) -> Any:
    """
    Get a file by its hashcode (filename in this implementation).

    Args:
        hashcode (str): The hashcode (filename) of the file.
        context (str): The directory path where the file is located.

    Returns:
        Any: A file object if found.

    Raises:
        FileNotFoundError: If the file is not found in the specified context.
    """
    if hashcode in os.listdir(context):
        return open(f"{context}/{hashcode}", "rb")
    else:
        raise FileNotFoundError(f"Couldn't find file {hashcode} in path {context}")

get_root_path()

Get the root path of the storage.

Returns:

Name Type Description
str str

The full path to the storage directory.
Source code in framework3/plugins/storage/local_storage.py 
77
78
79
80
81
82
83
84
def get_root_path(self) -> str:
    """
    Get the root path of the storage.

    Returns:
        str: The full path to the storage directory.
    """
    return self._base_path

list_stored_files(context)

List all files in the specified context.

Parameters:

Name Type Description Default
context str

The directory path to list files from.

 required

Returns:

Type Description
List[str]

List[str]: A list of file names in the specified context.
Source code in framework3/plugins/storage/local_storage.py 
111
112
113
114
115
116
117
118
119
120
121
def list_stored_files(self, context: str) -> List[str]:
    """
    List all files in the specified context.

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

    Returns:
        List[str]: A list of file names in the specified context.
    """
    return os.listdir(context)

upload_file(file, file_name, context, direct_stream=False)

Upload a file to the specified context.

Parameters:

Name Type Description Default
file Any

The file content to be uploaded.

 required
file_name str

The name of the file.

 required
context str

The directory path where the file will be saved.

 required
direct_stream bool

Not used in this implementation. Defaults to False.

 False

Returns:

Type Description
str | None

str | None: The file name if successful, None otherwise.
Source code in framework3/plugins/storage/local_storage.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def upload_file(
    self, file, file_name: str, context: str, direct_stream: bool = False
) -> str | None:
    """
    Upload a file to the specified context.

    Args:
        file (Any): The file content to be uploaded.
        file_name (str): The name of the file.
        context (str): The directory path where the file will be saved.
        direct_stream (bool, optional): Not used in this implementation. Defaults to False.

    Returns:
        str | None: The file name if successful, None otherwise.
    """
    try:
        Path(context).mkdir(parents=True, exist_ok=True)
        print(f"\t * Saving in local path: {context}/{file_name}")
        pickle.dump(file, open(f"{context}/{file_name}", "wb"))
        print("\t * Saved !")
        return file_name
    except Exception as ex:
        print(ex)
    return None