Files
notes/02-projects/Uplink/rclone_engine.py.md
T
2026-07-14 08:05:36 -07:00

246 lines
8.7 KiB
Markdown

## Summary
This file contains the primary working logic for backup and zipping. This file does the heavy lifting of the project.
## Classes
### Engine
```python
class RcloneEngine(BaseBackupEngine):
"""
This class orchestrates and controls all processes for backups using rclone
"""
def __init__(progress_handle = None) -> None:
"""
Creates an instance of the RcloneEngine class
args:
- progress_handle: rich progress handler. This is used to handle
displaying progress bars during the entire process. It is passed to all
classes as needed.
"""
def run(self, tasks: list[BackupTask], dry_run: bool = False): -> bool
"""
This function orchestrates and runs the zipping and backup process.
args:
- tasks (list[BackupTask]): list of BackgroundTask objects to be
processed.
- dry_run (bool, optional): Dry run flag. If this flag is set, this
function will return None. A list of all files to be compressed, their
total size, and the path to the output file will be printed to the
console, and logged. Defaults to False.
"""
```
#### run
This function orchestrates and runs the zipping and backup process.
args:
- tasks (list\[BackupTask\]): list of BackgroundTask objects to be processed.
- dry_run (bool, optional): Dry run flag. If this flag is set, this function will return None. A list of all files to be compressed, their total size, and the path to the output file will be printed to the console, and logged. Defaults to False.
### Compressor
```python
class Compressor:
"""
This class hanldes file compression operations.
"""
def __init__(progress_handle = None) -> None:
"""
Creates an instance of the Compressor class
args:
- progress_handle: rich progress handler. This is used to handle
displaying progress bars during the zip process
"""
def compress_directory(
self,
task: BackupTask,
dry_run: bool = False
) -> Path | None:
"""
This method orchestrates all functions related to file zipping. It calls
all required class methods and handles data flow and error handling for
compression tasks
args:
- task (BackupTask): BackupTask object current being worked on. This is
used to get all info regarding the zipping task
- dry_run (bool, optional): Dry run flag. If this flag is set, this
function will return None. A list of all files to be compressed, their
total size, and the path to the output file will be printed to the
console, and logged. Defaults to False.
returns:
- path: Path to output file of completed zip operation.
- None: In any cases where an output file is note written
raises:
- InvalidBackupTaskError: If provided BackupTask is invalid
- CompressionError: If any error is raised during the compression
process
"""
def _get_file_list(self, task: BackupTask) -> list[Path]:
"""
This method reads the source path from task and returns a list of all
files and directories to be compressed, ignoring any files or directories
that match exclude patterns in provided task object
args:
- task (BackupTask): BackupTask object current being worked on. This
is used to get the source directory and exclude patterns to select
files for compression
returns:
- list[Path]: List of paths to be compressed. If no files are found,
this list will be empty
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided source path is not found
"""
def _run_zip(self, files: list[Path]) -> path | None:
"""
This method handles the actual zip process. This is also handles
displaying the progress bar if one is desired.
args:
- files (list[Path]): list of files/paths to compress
returns:
- Path: path of output zip file.
- None: in cases where an output file could not be written
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided path is not found
- CompressionError: If any error is raised during the compression
process
"""
```
This class handles file compression of all directories and files in the source directory.
#### compress_directory
This method orchestrates all functions related to file zipping. It calls all required class methods and handles data flow and error handling for compression tasks
args:
- task (BackupTask): BackupTask object current being worked on. This is used to get all info regarding the zipping task
- dry_run (bool, optional): Dry run flag. If this flag is set, this function will return None. A list of all files to be compressed, their total size, and the path to the output file will be printed to the console, and logged. Defaults to False.
-
returns:
- Path: Path to output file of completed zip operation.
- None: In any cases where an output file is note written
-
raises:
- InvalidBackupTaskError: If provided BackupTask is invalid
- CompressionError: If any error is raised during the compression process
#### _get_file_list
This method reads the source path from task and returns a list of all files and directories to be compressed, ignoring any files or directories that match exclude patterns in provided task object
args:
- task (BackupTask): BackupTask object current being worked on. This is used to get the source directory and exclude patterns to select files for compression
returns:
- list\[Path\]: List of paths to be compressed. If no files are found, this list will be empty
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided source path is not found
#### _run_zip
This method handles the actual zip process. This is also handles displaying the progress bar if one is desired.
args:
- files (list\[Path\]): list of files/paths to compress
returns:
- Path: path of output zip file.
- None: in cases where an output file could not be written
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided path is not found
- CompressionError: If any error is raised during the compression process
### Backup
This class handles the backup process using rclone
```python
class Backup:
"""
This class handles the backup process using rclone
"""
def __init__(progress_handle = None) -> None:
"""
Creates an instance of the Uplink Class
args:
- progress_handle: rich progress handler. This is used to handle
displaying progress bars during the zip process
"""
def backup(
self,
source: list[Path] | Path,
dest: Path,
dry_rule: bool = False
) -> bool:
"""
Back ups source to dest using rclone
args:
- source (list[Path] | Path): Accepts either a list of Path objects
or a single Path object. This will be copied to the destination path
- dest (Path): Destination for source files to be backed up to.
- dry_run (bool, optional): Dry run flag. If this flag is set, this
function will return None. A list of all files to be compressed,
their
total size, and the path to the output file will be printed to the
console, and logged. Defaults to False.
returns:
- bool: True if backup process completes without errors. False
otherwise.
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided source path or paths are not found
- BackupError: If any errors happen during the backup process
"""
@contextmanager
def _create_from_file(self, source: list[Path]) -> Iterator[str]:
"""
Context manager for temp rclone from file. Contains list of paths to be
backed up by rclone.
args:
- source (list[Path]): List of absolute paths to be backed up
yields:
- Iterator[str]: Absolute path of the temp file container list of
source paths.
raises:
- IOError: If unable to write tmp file, or if any other IO errors are
encounted
"""
```
#### backup
Back ups source to dest using rclone
args:
- source (list[Path] | Path): Accepts either a list of Path objects or a single Path object. This will be copied to the destination path
- dest (Path): Destination for source files to be backed up to.
- dry_run (bool, optional): Dry run flag. If this flag is set, this function will return None. A list of all files to be compressed, their total size, and the path to the output file will be printed to the console, and logged. Defaults to False.
returns:
- bool: True if backup process completes without errors. False otherwise.
raises:
- IOError: If any IO errors are encounted
- FileNotFoundError: If provided source path or paths are not found
- BackupError: If any errors happen during the backup process
#### _create_from_file
Context manager for temp rclone from file. Contains list of paths to be backed up by rclone.
args:
- source (list[Path]): List of absolute paths to be backed up
yields:
- Iterator[str]: Absolute path of the temp file container list of source paths.
raises:
- IOError: If unable to write tmp file, or if any other IO errors are
encounted