backup: method ordering

This commit is contained in:
2026-04-12 18:00:32 +03:00
parent b90b87caa1
commit 7e1a8e2e99
+57 -57
View File
@@ -296,7 +296,62 @@ class BackupManager:
self.storages = storages self.storages = storages
self.notifiers = notifiers 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""" """Run backup script as the specified user"""
try: try:
logger.info(f"Running backup script {script_path} (user {username})") logger.info(f"Running backup script {script_path} (user {username})")
@@ -335,7 +390,7 @@ class BackupManager:
self.errors.append(f"App {username}: {error_msg}") self.errors.append(f"App {username}: {error_msg}")
return False return False
def send_notification(self, success: bool) -> None: def _send_notification(self, success: bool) -> None:
"""Send notification to Notifiers""" """Send notification to Notifiers"""
host = self.config.host_name host = self.config.host_name
@@ -366,61 +421,6 @@ class BackupManager:
except Exception as e: except Exception as e:
logger.error(f"Failed to send notification: {str(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( def initialize(
config_path: Path, config_path: Path,