From a217c79e7d916c9542e3c16ed3a5c8e2ba558848 Mon Sep 17 00:00:00 2001 From: Anton Vakhrushev Date: Sat, 20 Dec 2025 21:03:32 +0300 Subject: [PATCH] Backup: extract restic storage into separate class --- files/backups/backup-all.py | 334 ++++++++++++++++++------------------ 1 file changed, 170 insertions(+), 164 deletions(-) diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py index 0ad787f..18880bf 100644 --- a/files/backups/backup-all.py +++ b/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)