Backups: add restic errors

files/backups/backup-all.py

@@ -6,18 +6,19 @@ then creates restic backups and sends notifications.
 """
 
 import itertools
-import os
-import sys
-import subprocess
 import logging
+import os
 import pwd
+import subprocess
+import sys
 import time
+import tomllib
 from abc import ABC
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Dict, List, Optional, Any
+from typing import Any, Dict, List, Optional
+
 import requests
-import tomllib
 
 # Default config path
 CONFIG_PATH = Path("/etc/backup/config.toml")
@@ -54,6 +55,12 @@ class Application:
     backup_targets: List[Path]
 
 
+@dataclass
+class BackupResult:
+    success: bool
+    error: Optional[str] = None
+
+
 @dataclass
 class StorageRunResult:
     name: str
@@ -76,7 +83,7 @@ def format_duration(seconds: float) -> str:
 class Storage(ABC):
     name: str
 
-    def backup(self, backup_dirs: List[str]) -> bool:
+    def backup(self, backup_dirs: List[str]) -> BackupResult:
         """Backup directories"""
         raise NotImplementedError()
 
@@ -101,17 +108,17 @@ class ResticStorage(Storage):
                 f"Missing storage configuration values for backend ResticStorage: '{self.name}'"
             )
 
-    def backup(self, backup_dirs: List[str]) -> bool:
+    def backup(self, backup_dirs: List[str]) -> BackupResult:
         if not backup_dirs:
             logger.warning("No backup directories found")
-            return True
+            return BackupResult(success=True)
         try:
             return self.__backup_internal(backup_dirs)
         except Exception as exc:  # noqa: BLE001
             logger.error("Restic backup process failed: %s", exc)
-            return False
+            return BackupResult(success=False, error=str(exc))
 
-    def __backup_internal(self, backup_dirs: List[str]) -> bool:
+    def __backup_internal(self, backup_dirs: List[str]) -> BackupResult:
         logger.info("Starting restic backup for storage '%s'", self.name)
         logger.info("Destination: %s", self.restic_repository)
 
@@ -120,25 +127,13 @@ class ResticStorage(Storage):
         env["RESTIC_PASSWORD"] = self.restic_password
         env.update(self.env)
 
-        backup_cmd = ["restic", "backup", "--verbose"] + backup_dirs
-        result = subprocess.run(backup_cmd, env=env, capture_output=True, text=True)
-
-        if result.returncode != 0:
-            logger.error("Restic backup failed: %s", result.stderr)
-            return False
-
-        logger.info("Restic backup completed successfully")
-
         check_cmd = ["restic", "check"]
-        result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
+        steps = [
-
+            ("backup", ["restic", "backup", "--verbose"] + backup_dirs),
-        if result.returncode != 0:
+            ("check", check_cmd),
-            logger.error("Restic check failed: %s", result.stderr)
+            (
-            return False
+                "forget/prune",
-
+                [
-        logger.info("Restic check completed successfully")
-
-        forget_cmd = [
                     "restic",
                     "forget",
                     "--compact",
@@ -147,23 +142,31 @@ class ResticStorage(Storage):
                     "90",
                     "--keep-monthly",
                     "36",
+                ],
+            ),
+            ("final check", check_cmd),
         ]
-        result = subprocess.run(forget_cmd, env=env, capture_output=True, text=True)
+
+        for step, cmd in steps:
+            error = self.__run_step(step, cmd, env)
+            if error is not None:
+                return BackupResult(success=False, error=f"restic {step}: {error}")
+
+        return BackupResult(success=True)
+
+    def __run_step(
+        self, step: str, cmd: List[str], env: Dict[str, str]
+    ) -> Optional[str]:
+        """Run a single restic command. Return None on success or error text."""
+        result = subprocess.run(cmd, env=env, capture_output=True, text=True)
 
         if result.returncode != 0:
-            logger.error("Restic forget/prune failed: %s", result.stderr)
+            error = result.stderr.strip() or result.stdout.strip() or "no output"
-            return False
+            logger.error("Restic %s failed: %s", step, error)
+            return error
 
-        logger.info("Restic forget/prune completed successfully")
+        logger.info("Restic %s completed successfully", step)
-
+        return None
-        result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
-
-        if result.returncode != 0:
-            logger.error("Final restic check failed: %s", result.stderr)
-            return False
-
-        logger.info("Final restic check completed successfully")
-        return True
 
 
 class Notifier(ABC):
@@ -357,12 +360,12 @@ class BackupManager:
                 logger.error(
                     "Storage '%s' raised an unexpected error: %s", storage.name, exc
                 )
-                backup_result = False
+                backup_result = BackupResult(success=False, error=str(exc))
             storage_duration = time.monotonic() - storage_start
             self.storage_results.append(
                 StorageRunResult(
                     name=storage.name,
-                    success=backup_result,
+                    success=backup_result.success,
                     duration=storage_duration,
                 )
             )
@@ -370,13 +373,16 @@ class BackupManager:
                 "Storage '%s' finished in %s (success=%s)",
                 storage.name,
                 format_duration(storage_duration),
-                backup_result,
+                backup_result.success,
             )
-            if not backup_result:
+            if not backup_result.success:
-                self.errors.append(f"Storage '{storage.name}' backup failed")
+                error_msg = f"Storage '{storage.name}' backup failed"
+                if backup_result.error:
+                    error_msg += f": {backup_result.error}"
+                self.errors.append(error_msg)
 
             # Determine overall success
-            overall_success = overall_success and backup_result
+            overall_success = overall_success and backup_result.success
 
         # Send notification
         self._send_notification(overall_success)