From 7e1a8e2e99e42603eaa4d30d7dbd451e97d6b78a Mon Sep 17 00:00:00 2001 From: Anton Vakhrushev Date: Sun, 12 Apr 2026 18:00:32 +0300 Subject: [PATCH] backup: method ordering --- files/backups/backup-all.py | 114 ++++++++++++++++++------------------ 1 file changed, 57 insertions(+), 57 deletions(-) diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py index 7f84d6d..e272c7b 100644 --- a/files/backups/backup-all.py +++ b/files/backups/backup-all.py @@ -296,7 +296,62 @@ class BackupManager: self.storages = storages self.notifiers = notifiers - def run_app_backup(self, script_path: str, app_dir: str, username: str) -> bool: + def run_backup_process(self, applications: List[Application]) -> bool: + """Main backup process""" + logger.info("Starting backup process") + logger.info(f"Found {len(applications)} application directories") + + # Process each user's backup + for app in applications: + app_dir = str(app.path) + username = app.owner + logger.info(f"Processing backup for app: {app_dir} (user {username})") + + if app.backup_script is None: + warning_msg = ( + f"No backup script found for app: {app_dir} (user {username})" + ) + logger.warning(warning_msg) + self.warnings.append(warning_msg) + continue + + self._run_app_backup(str(app.backup_script), app_dir, username) + + # Collect backup directories from applications + backup_dirs: List[str] = [] + for app in applications: + for target in app.backup_targets: + target_str = str(target) + if target_str not in backup_dirs: + backup_dirs.append(target_str) + logger.info(f"Found backup directories: {backup_dirs}") + + overall_success = True + + for storage in self.storages: + backup_result = storage.backup(backup_dirs) + if not backup_result: + self.errors.append("Restic backup failed") + + # Determine overall success + overall_success = overall_success and backup_result + + # Send notification + self._send_notification(overall_success) + + logger.info("Backup process completed") + + if self.errors: + logger.error(f"Backup completed with {len(self.errors)} errors") + return False + elif self.warnings: + logger.warning(f"Backup completed with {len(self.warnings)} warnings") + return True + else: + logger.info("Backup completed successfully") + return True + + def _run_app_backup(self, script_path: str, app_dir: str, username: str) -> bool: """Run backup script as the specified user""" try: logger.info(f"Running backup script {script_path} (user {username})") @@ -335,7 +390,7 @@ class BackupManager: self.errors.append(f"App {username}: {error_msg}") return False - def send_notification(self, success: bool) -> None: + def _send_notification(self, success: bool) -> None: """Send notification to Notifiers""" host = self.config.host_name @@ -366,61 +421,6 @@ class BackupManager: except Exception as e: logger.error(f"Failed to send notification: {str(e)}") - def run_backup_process(self, applications: List[Application]) -> bool: - """Main backup process""" - logger.info("Starting backup process") - logger.info(f"Found {len(applications)} application directories") - - # Process each user's backup - for app in applications: - app_dir = str(app.path) - username = app.owner - logger.info(f"Processing backup for app: {app_dir} (user {username})") - - if app.backup_script is None: - warning_msg = ( - f"No backup script found for app: {app_dir} (user {username})" - ) - logger.warning(warning_msg) - self.warnings.append(warning_msg) - continue - - self.run_app_backup(str(app.backup_script), app_dir, username) - - # Collect backup directories from applications - backup_dirs: List[str] = [] - for app in applications: - for target in app.backup_targets: - target_str = str(target) - if target_str not in backup_dirs: - backup_dirs.append(target_str) - logger.info(f"Found backup directories: {backup_dirs}") - - overall_success = True - - for storage in self.storages: - backup_result = storage.backup(backup_dirs) - if not backup_result: - self.errors.append("Restic backup failed") - - # Determine overall success - overall_success = overall_success and backup_result - - # Send notification - self.send_notification(overall_success) - - logger.info("Backup process completed") - - if self.errors: - logger.error(f"Backup completed with {len(self.errors)} errors") - return False - elif self.warnings: - logger.warning(f"Backup completed with {len(self.warnings)} warnings") - return True - else: - logger.info("Backup completed successfully") - return True - def initialize( config_path: Path,