Backup: extract restic storage into separate class

files/backups/backup-all.py

@@ -15,7 +15,6 @@ from pathlib import Path
 from typing import Dict, List, Optional
 import requests
 import tomllib
-from collections.abc import Iterable
 
 
 # Configure logging
@@ -55,78 +54,80 @@ class Config:
     notifications: Dict[str, TelegramConfig]
 
 
-def read_config(config_path: Path) -> Config:
-    try:
-        with config_path.open("rb") as config_file:
-            raw_config = tomllib.load(config_file)
-    except OSError as e:
-        logger.error(f"Failed to read config file {config_path}: {e}")
-        raise
+class ResticStorage:
+    def __init__(self, cfg: StorageConfig):
+        if cfg.type != "restic":
+            raise ValueError(f"Unsupported storage type for ResticStorage: {cfg.type}")
+        self.cfg = cfg
 
-    roots_raw = raw_config.get("roots") or []
-    if not isinstance(roots_raw, list) or not roots_raw:
-        raise ValueError("roots must be a non-empty list of paths in config.toml")
-    roots = [Path(root) for root in roots_raw]
+    def backup(self, backup_dirs: List[str]) -> bool:
+        if not backup_dirs:
+            logger.warning("No backup directories found")
+            return True
 
-    storage_raw = raw_config.get("storage") or {}
-    storage: Dict[str, StorageConfig] = {}
-    for name, cfg in storage_raw.items():
-        if not isinstance(cfg, dict):
-            raise ValueError(f"Storage config for {name} must be a table")
-        storage[name] = StorageConfig(
-            type=cfg.get("type", ""),
-            restic_repository=cfg.get("restic_repository", ""),
-            restic_password=cfg.get("restic_password", ""),
-            aws_access_key_id=cfg.get("aws_access_key_id", ""),
-            aws_secret_access_key=cfg.get("aws_secret_access_key", ""),
-            aws_default_region=cfg.get("aws_default_region", ""),
-        )
+        try:
+            logger.info("Starting restic backup")
+            logger.info("Destination: %s", self.cfg.restic_repository)
 
-    if not storage:
-        raise ValueError("At least one storage backend must be configured")
-
-    notifications_raw = raw_config.get("notifications") or {}
-    notifications: Dict[str, TelegramConfig] = {}
-    for name, cfg in notifications_raw.items():
-        if not isinstance(cfg, dict):
-            raise ValueError(f"Notification config for {name} must be a table")
-        notifications[name] = TelegramConfig(
-            type=cfg.get("type", ""),
-            telegram_bot_token=cfg.get("telegram_bot_token", ""),
-            telegram_chat_id=cfg.get("telegram_chat_id", ""),
-            notifications_name=cfg.get("notifications_name", ""),
-        )
-
-    if not notifications:
-        raise ValueError("At least one notification backend must be configured")
-
-    for name, cfg in storage.items():
-        if not all(
-            [
-                cfg.type,
-                cfg.restic_repository,
-                cfg.restic_password,
-                cfg.aws_access_key_id,
-                cfg.aws_secret_access_key,
-                cfg.aws_default_region,
-            ]
-        ):
-            raise ValueError(f"Missing storage configuration values for backend {name}")
-
-    for name, cfg in notifications.items():
-        if not all(
-            [
-                cfg.type,
-                cfg.telegram_bot_token,
-                cfg.telegram_chat_id,
-                cfg.notifications_name,
-            ]
-        ):
-            raise ValueError(
-                f"Missing notification configuration values for backend {name}"
+            env = os.environ.copy()
+            env.update(
+                {
+                    "RESTIC_REPOSITORY": self.cfg.restic_repository,
+                    "RESTIC_PASSWORD": self.cfg.restic_password,
+                    "AWS_ACCESS_KEY_ID": self.cfg.aws_access_key_id,
+                    "AWS_SECRET_ACCESS_KEY": self.cfg.aws_secret_access_key,
+                    "AWS_DEFAULT_REGION": self.cfg.aws_default_region,
+                }
             )
 
-    return Config(roots=roots, storage=storage, notifications=notifications)
+            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)
+
+            if result.returncode != 0:
+                logger.error("Restic check failed: %s", result.stderr)
+                return False
+
+            logger.info("Restic check completed successfully")
+
+            forget_cmd = [
+                "restic",
+                "forget",
+                "--compact",
+                "--prune",
+                "--keep-daily",
+                "90",
+                "--keep-monthly",
+                "36",
+            ]
+            result = subprocess.run(forget_cmd, env=env, capture_output=True, text=True)
+
+            if result.returncode != 0:
+                logger.error("Restic forget/prune failed: %s", result.stderr)
+                return False
+
+            logger.info("Restic forget/prune completed successfully")
+
+            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
+
+        except Exception as exc:  # noqa: BLE001
+            logger.error("Restic backup process failed: %s", exc)
+            return False
 
 
 CONFIG_PATH = Path("/etc/backup/config.toml")
@@ -146,16 +147,12 @@ class Application:
 
 
 class BackupManager:
-    def __init__(self):
+    def __init__(self, config: Config, storage: ResticStorage):
         self.errors: List[str] = []
         self.warnings: List[str] = []
         self.successful_backups: List[str] = []
-        self.config = read_config(CONFIG_PATH)
-
-    def _select_storage(self) -> StorageConfig:
-        if "yandex" in self.config.storage:
-            return self.config.storage["yandex"]
-        return next(iter(self.config.storage.values()))
+        self.config = config
+        self.storage = storage
 
     def _select_telegram(self) -> Optional[TelegramConfig]:
         if "telegram" in self.config.notifications:
@@ -296,93 +293,6 @@ class BackupManager:
 
         return backup_dirs
 
-    def run_restic_backup(self, backup_dirs: List[str]) -> bool:
-        """Run restic backup for all backup directories"""
-        if not backup_dirs:
-            logger.warning("No backup directories found")
-            return True
-
-        storage_cfg = self._select_storage()
-
-        try:
-            logger.info("Starting restic backup")
-            logger.info("Destination: %s", storage_cfg.restic_repository)
-
-            # Set environment variables for restic
-            env = os.environ.copy()
-            env.update(
-                {
-                    "RESTIC_REPOSITORY": storage_cfg.restic_repository,
-                    "RESTIC_PASSWORD": storage_cfg.restic_password,
-                    "AWS_ACCESS_KEY_ID": storage_cfg.aws_access_key_id,
-                    "AWS_SECRET_ACCESS_KEY": storage_cfg.aws_secret_access_key,
-                    "AWS_DEFAULT_REGION": storage_cfg.aws_default_region,
-                }
-            )
-
-            # Run backup
-            backup_cmd = ["restic", "backup", "--verbose"] + backup_dirs
-            result = subprocess.run(backup_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                error_msg = f"Restic backup failed: {result.stderr}"
-                logger.error(error_msg)
-                self.errors.append(f"Restic backup: {error_msg}")
-                return False
-
-            logger.info("Restic backup completed successfully")
-
-            # Run check
-            check_cmd = ["restic", "check"]
-            result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                error_msg = f"Restic check failed: {result.stderr}"
-                logger.error(error_msg)
-                self.errors.append(f"Restic check: {error_msg}")
-                return False
-
-            logger.info("Restic check completed successfully")
-
-            # Run forget and prune
-            forget_cmd = [
-                "restic",
-                "forget",
-                "--compact",
-                "--prune",
-                "--keep-daily",
-                "90",
-                "--keep-monthly",
-                "36",
-            ]
-            result = subprocess.run(forget_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                error_msg = f"Restic forget/prune failed: {result.stderr}"
-                logger.error(error_msg)
-                self.errors.append(f"Restic forget/prune: {error_msg}")
-                return False
-
-            logger.info("Restic forget/prune completed successfully")
-
-            # Final check
-            result = subprocess.run(check_cmd, env=env, capture_output=True, text=True)
-
-            if result.returncode != 0:
-                error_msg = f"Final restic check failed: {result.stderr}"
-                logger.error(error_msg)
-                self.errors.append(f"Final restic check: {error_msg}")
-                return False
-
-            logger.info("Final restic check completed successfully")
-            return True
-
-        except Exception as e:
-            error_msg = f"Restic backup process failed: {str(e)}"
-            logger.error(error_msg)
-            self.errors.append(f"Restic: {error_msg}")
-            return False
-
     def send_telegram_notification(self, success: bool) -> None:
         """Send notification to Telegram"""
         telegram_cfg = self._select_telegram()
@@ -464,7 +374,10 @@ class BackupManager:
         logger.info(f"Found backup directories: {backup_dirs}")
 
         # Run restic backup
-        restic_success = self.run_restic_backup(backup_dirs)
+        restic_success = self.storage.backup(backup_dirs)
+
+        if not restic_success:
+            self.errors.append("Restic backup failed")
 
         # Determine overall success
         overall_success = restic_success and len(self.errors) == 0
@@ -485,9 +398,102 @@ class BackupManager:
             return True
 
 
+def initialize(config_path: Path) -> BackupManager:
+    try:
+        with config_path.open("rb") as config_file:
+            raw_config = tomllib.load(config_file)
+    except OSError as e:
+        logger.error(f"Failed to read config file {config_path}: {e}")
+        raise
+
+    roots_raw = raw_config.get("roots") or []
+    if not isinstance(roots_raw, list) or not roots_raw:
+        raise ValueError("roots must be a non-empty list of paths in config.toml")
+    roots = [Path(root) for root in roots_raw]
+
+    storage_raw = raw_config.get("storage") or {}
+    storage: Dict[str, StorageConfig] = {}
+    for name, cfg in storage_raw.items():
+        if not isinstance(cfg, dict):
+            raise ValueError(f"Storage config for {name} must be a table")
+        storage[name] = StorageConfig(
+            type=cfg.get("type", ""),
+            restic_repository=cfg.get("restic_repository", ""),
+            restic_password=cfg.get("restic_password", ""),
+            aws_access_key_id=cfg.get("aws_access_key_id", ""),
+            aws_secret_access_key=cfg.get("aws_secret_access_key", ""),
+            aws_default_region=cfg.get("aws_default_region", ""),
+        )
+
+    if not storage:
+        raise ValueError("At least one storage backend must be configured")
+
+    notifications_raw = raw_config.get("notifications") or {}
+    notifications: Dict[str, TelegramConfig] = {}
+    for name, cfg in notifications_raw.items():
+        if not isinstance(cfg, dict):
+            raise ValueError(f"Notification config for {name} must be a table")
+        notifications[name] = TelegramConfig(
+            type=cfg.get("type", ""),
+            telegram_bot_token=cfg.get("telegram_bot_token", ""),
+            telegram_chat_id=cfg.get("telegram_chat_id", ""),
+            notifications_name=cfg.get("notifications_name", ""),
+        )
+
+    if not notifications:
+        raise ValueError("At least one notification backend must be configured")
+
+    for name, cfg in storage.items():
+        if not all(
+            [
+                cfg.type,
+                cfg.restic_repository,
+                cfg.restic_password,
+                cfg.aws_access_key_id,
+                cfg.aws_secret_access_key,
+                cfg.aws_default_region,
+            ]
+        ):
+            raise ValueError(f"Missing storage configuration values for backend {name}")
+
+    for name, cfg in notifications.items():
+        if not all(
+            [
+                cfg.type,
+                cfg.telegram_bot_token,
+                cfg.telegram_chat_id,
+                cfg.notifications_name,
+            ]
+        ):
+            raise ValueError(
+                f"Missing notification configuration values for backend {name}"
+            )
+
+    config = Config(roots=roots, storage=storage, notifications=notifications)
+    storage_backend = _create_storage(config)
+
+    return BackupManager(config=config, storage=storage_backend)
+
+
+def _create_storage(config: Config) -> ResticStorage:
+    # Prefer explicit yandex key if present
+    if "yandex" in config.storage:
+        candidate = config.storage["yandex"]
+        if candidate.type != "restic":
+            raise ValueError("Storage 'yandex' must be of type 'restic'")
+        return ResticStorage(candidate)
+
+    # Otherwise take the first restic storage
+    for name, cfg in config.storage.items():
+        if cfg.type == "restic":
+            return ResticStorage(cfg)
+
+    raise ValueError("No restic storage backend configured")
+
+
 def main():
     try:
-        backup_manager = BackupManager()
+        backup_manager = initialize(CONFIG_PATH)
         success = backup_manager.run_backup_process()
         if not success:
             sys.exit(1)