diff --git a/files/backups/backup-all.py b/files/backups/backup-all.py index b1d2df2..c08154a 100644 --- a/files/backups/backup-all.py +++ b/files/backups/backup-all.py @@ -10,12 +10,14 @@ import sys import subprocess import logging import pwd +from abc import ABC from dataclasses import dataclass from pathlib import Path -from typing import Dict, List, Optional +from typing import Dict, List, Optional, Any import requests import tomllib +# Default config path CONFIG_PATH = Path("/etc/backup/config.toml") # File name to store directories and files to back up @@ -37,16 +39,6 @@ logging.basicConfig( logger = logging.getLogger(__name__) -@dataclass -class StorageConfig: - type: str - restic_repository: str - restic_password: str - aws_access_key_id: str - aws_secret_access_key: str - aws_default_region: str - - @dataclass class TelegramConfig: type: str @@ -58,7 +50,6 @@ class TelegramConfig: @dataclass class Config: roots: List[Path] - storage: Dict[str, StorageConfig] notifications: Dict[str, TelegramConfig] @@ -68,11 +59,35 @@ class Application: owner: str -class ResticStorage: - def __init__(self, cfg: StorageConfig): - if cfg.type != "restic": - raise ValueError(f"Unsupported storage type for ResticStorage: {cfg.type}") - self.cfg = cfg +class Storage(ABC): + def backup(self, backup_dirs: List[str]) -> bool: + """Backup directories""" + raise NotImplementedError() + + +class ResticStorage(Storage): + TYPE_NAME = "restic" + + def __init__(self, name: str, params: Dict[str, Any]): + self.name = name + self.restic_repository = str(params.get("restic_repository", "")) + self.restic_password = str(params.get("restic_password", "")) + self.aws_access_key_id = str(params.get("aws_access_key_id", "")) + self.aws_secret_access_key = str(params.get("aws_secret_access_key", "")) + self.aws_default_region = str(params.get("aws_default_region", "")) + + if not all( + [ + self.restic_repository, + self.restic_password, + self.aws_access_key_id, + self.aws_secret_access_key, + self.aws_default_region, + ] + ): + raise ValueError( + f"Missing storage configuration values for backend ResticStorage: '{self.name}'" + ) def backup(self, backup_dirs: List[str]) -> bool: if not backup_dirs: @@ -81,16 +96,16 @@ class ResticStorage: try: logger.info("Starting restic backup") - logger.info("Destination: %s", self.cfg.restic_repository) + logger.info("Destination: %s", self.restic_repository) 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, + "RESTIC_REPOSITORY": self.restic_repository, + "RESTIC_PASSWORD": self.restic_password, + "AWS_ACCESS_KEY_ID": self.aws_access_key_id, + "AWS_SECRET_ACCESS_KEY": self.aws_secret_access_key, + "AWS_DEFAULT_REGION": self.aws_default_region, } ) @@ -145,12 +160,12 @@ class ResticStorage: class BackupManager: - def __init__(self, config: Config, storage: ResticStorage): + def __init__(self, config: Config, storages: List[Storage]): self.errors: List[str] = [] self.warnings: List[str] = [] self.successful_backups: List[str] = [] self.config = config - self.storage = storage + self.storages = storages def _select_telegram(self) -> Optional[TelegramConfig]: if "telegram" in self.config.notifications: @@ -371,14 +386,15 @@ class BackupManager: backup_dirs = self.get_backup_directories() logger.info(f"Found backup directories: {backup_dirs}") - # Run restic backup - restic_success = self.storage.backup(backup_dirs) + overall_success = True - if not restic_success: - self.errors.append("Restic backup failed") + 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 = restic_success and len(self.errors) == 0 + # Determine overall success + overall_success = overall_success and backup_result # Send notification self.send_telegram_notification(overall_success) @@ -410,20 +426,14 @@ def initialize(config_path: Path) -> BackupManager: 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): + storages: List[Storage] = [] + for name, params in storage_raw.items(): + if not isinstance(params, 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: + storage_type = params.get("type", "") + if storage_type == ResticStorage.TYPE_NAME: + storages.append(ResticStorage(name, params)) + if not storages: raise ValueError("At least one storage backend must be configured") notifications_raw = raw_config.get("notifications") or {} @@ -441,19 +451,6 @@ def initialize(config_path: Path) -> BackupManager: 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( [ @@ -467,26 +464,9 @@ def initialize(config_path: Path) -> BackupManager: f"Missing notification configuration values for backend {name}" ) - config = Config(roots=roots, storage=storage, notifications=notifications) - storage_backend = _create_storage(config) + config = Config(roots=roots, notifications=notifications) - 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") + return BackupManager(config=config, storages=storages) def main():