File size: 7,476 Bytes
ad3bfb6 |
1 2 3 4 5 6 7 8 9 10 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 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 |
"""
file_manager.py
This module provides the FileManager class, a high-level interface for managing
files and directories inside a specified media folder. It centralizes common
operations such as reading, writing, listing, copying, moving, and deleting files,
ensuring that all actions are safely constrained within the defined root directory.
The class is designed to simplify file handling logic in larger applications by
offering a consistent, validated, and extensible API for filesystem interactions.
"""
from pathlib import Path
class FileManager:
"""
FileManager is a utility class that encapsulates common filesystem operations
within a defined media directory. It ensures that all file manipulations remain
inside the configured root folder and provides helper methods to interact with
files and subdirectories in a structured, safe, and predictable manner.
Typical use cases include managing uploaded media, performing batch operations
on directory contents, and abstracting filesystem complexity behind a clean API.
"""
def __init__(self, media_folder: Path):
"""
Initialize the FileManager with a specific root directory for all file
operations.
Parameters
----------
media_folder : Path
The base directory where all file and folder operations will be performed.
It must be a valid filesystem path. If the directory does not exist, the
instance will attempt to create it automatically.
Raises
------
ValueError
If the provided media_folder path is not a valid directory or cannot be
created.
"""
self.media_folder = Path(media_folder)
if not self.media_folder.exists():
try:
self.media_folder.mkdir(parents=True)
except Exception as exc:
raise ValueError(
f"Unable to create media folder at: {self.media_folder}"
) from exc
if not self.media_folder.is_dir():
raise ValueError(f"Media folder is not a directory: {self.media_folder}")
def upload_file(self, file_handler, destination: Path):
"""
Upload a file to a target destination inside the media folder.
This method takes a file-like object and writes its contents to the
specified destination within the media folder. The method ensures the path
remains inside the media folder and creates necessary directories. It
returns a structured response indicating whether the operation succeeded
or failed, along with any relevant error message.
Parameters
----------
file_handler : file-like object
A file-like object opened in binary mode that provides a `.read()` method.
destination : Path
The relative or absolute path (within the media folder) where the file
should be saved.
Returns
-------
dict
A dictionary with:
- "operation_success" (bool): True if the file was saved successfully.
- "error" (str): An empty string on success, or the error message on failure.
Raises
------
None
Any exceptions are captured and returned inside the result dictionary.
"""
try:
destination = self.media_folder / destination
destination = destination.resolve()
# Ensure the destination lies inside the media folder
if self.media_folder not in destination.parents:
return {
"operation_success": False,
"error": "Destination path is outside the media folder."
}
# Create parent directories if needed
destination.parent.mkdir(parents=True, exist_ok=True)
with open(destination, "wb") as f:
f.write(file_handler.read())
return {"operation_success": True, "error": ""}
except Exception as exc:
return {
"operation_success": False,
"error": str(exc)
}
def get_file(self, file_path: Path):
"""
Retrieve a file inside the media folder and return an open file handler.
This method receives a path pointing to a file expected to be located
inside the media folder. If the file exists and is valid, the method
returns a file handler opened in binary read mode. If the file does not
exist or the resolved path escapes the media folder, the method returns
None.
Parameters
----------
file_path : Path
The relative or absolute path to the file within the media folder.
Returns
-------
file object or None
A file handler opened in 'rb' mode if the file exists and is accessible.
Returns None if the file does not exist or the path is invalid.
Raises
------
None
All exceptions are handled internally and result in returning None.
"""
try:
target = self.media_folder / file_path
target = target.resolve()
# Ensure the resolved path is inside the media folder
if self.media_folder not in target.parents:
return None
if not target.exists() or not target.is_file():
return None
return open(target, "rb")
except Exception:
return None
def compute_sha1(self, file_handler):
"""
Compute the SHA-1 checksum of a file handler.
This method calculates the SHA-1 hash of the content provided by a
file-like object opened in binary mode. The method preserves the
current file pointer position by saving it, rewinding to the start
of the file, computing the hash, and restoring the file pointer to
its original position.
Parameters
----------
file_handler : file-like object
A file-like object opened in binary mode, providing `.read()` and
`.seek()` methods.
Returns
-------
str
The hexadecimal SHA-1 checksum of the file content.
Raises
------
ValueError
If the provided file handler is invalid or does not support the
required read/seek operations.
"""
import hashlib
if not hasattr(file_handler, "read") or not hasattr(file_handler, "seek"):
raise ValueError("The provided file handler does not support read/seek operations.")
original_pos = file_handler.tell()
try:
file_handler.seek(0)
sha1 = hashlib.sha1()
for chunk in iter(lambda: file_handler.read(8192), b""):
sha1.update(chunk)
file_handler.seek(original_pos)
return sha1.hexdigest()
except Exception as exc:
try:
file_handler.seek(original_pos)
except Exception:
pass
raise ValueError(f"Unable to compute SHA-1 checksum: {exc}") from exc
|