Backup: parse config to dataclasses

files/backups/backup-all.py

@@ -12,7 +12,7 @@ import logging
 import pwd
 from dataclasses import dataclass
 from pathlib import Path
-from typing import List, Optional
+from typing import Dict, List, Optional
 import requests
 import tomllib
 from collections.abc import Iterable
@@ -29,43 +29,107 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
-try:
-    with open("/etc/backup/config.toml", "rb") as config_file:
-        config = tomllib.load(config_file)
-except OSError as e:
-    logger.error(f"Failed to read config file: {e}")
-    raise
 
-ROOTS = config.get("roots") or []
+@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
 
-storage_cfg = config.get("storage", {}).get("yandex", {})
-RESTIC_REPOSITORY = storage_cfg.get("restic_repository")
-RESTIC_PASSWORD = storage_cfg.get("restic_password")
-AWS_ACCESS_KEY_ID = storage_cfg.get("aws_access_key_id")
-AWS_SECRET_ACCESS_KEY = storage_cfg.get("aws_secret_access_key")
-AWS_DEFAULT_REGION = storage_cfg.get("aws_default_region")
 
-notifications_cfg = config.get("notifications", {}).get("telegram", {})
-TELEGRAM_BOT_TOKEN = notifications_cfg.get("telegram_bot_token")
-TELEGRAM_CHAT_ID = notifications_cfg.get("telegram_chat_id")
-NOTIFICATIONS_NAME = notifications_cfg.get("notifications_name")
+@dataclass
+class TelegramConfig:
+    type: str
+    telegram_bot_token: str
+    telegram_chat_id: str
+    notifications_name: str
 
-if not isinstance(ROOTS, Iterable) or not ROOTS:
-    raise ValueError("roots must be a non-empty list of paths in config.toml")
 
-if not all(
-    [
-        RESTIC_REPOSITORY,
-        RESTIC_PASSWORD,
-        AWS_ACCESS_KEY_ID,
-        AWS_SECRET_ACCESS_KEY,
-        AWS_DEFAULT_REGION,
-        TELEGRAM_BOT_TOKEN,
-        TELEGRAM_CHAT_ID,
-        NOTIFICATIONS_NAME,
-    ]
-):
-    raise ValueError("Missing required configuration values in config.toml")
+@dataclass
+class Config:
+    roots: List[Path]
+    storage: Dict[str, StorageConfig]
+    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
+
+    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}"
+            )
+
+    return Config(roots=roots, storage=storage, notifications=notifications)
+
+
+CONFIG_PATH = Path("/etc/backup/config.toml")
 
 # File name to store directories and files to back up
 BACKUP_TARGETS_FILE = "backup-targets"
@@ -86,11 +150,22 @@ class BackupManager:
         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()))
+
+    def _select_telegram(self) -> Optional[TelegramConfig]:
+        if "telegram" in self.config.notifications:
+            return self.config.notifications["telegram"]
+        return next(iter(self.config.notifications.values()), None)
 
     def find_applications(self) -> List[Application]:
         """Get all application directories and their owners."""
         applications: List[Application] = []
-        source_dirs = itertools.chain(*(Path(root).iterdir() for root in ROOTS))
+        source_dirs = itertools.chain(*(root.iterdir() for root in self.config.roots))
 
         for app_dir in source_dirs:
             if "lost+found" in str(app_dir):
@@ -227,19 +302,21 @@ class BackupManager:
             logger.warning("No backup directories found")
             return True
 
+        storage_cfg = self._select_storage()
+
         try:
             logger.info("Starting restic backup")
-            logger.info("Destination: %s", RESTIC_REPOSITORY)
+            logger.info("Destination: %s", storage_cfg.restic_repository)
 
             # Set environment variables for restic
             env = os.environ.copy()
             env.update(
                 {
-                    "RESTIC_REPOSITORY": RESTIC_REPOSITORY,
-                    "RESTIC_PASSWORD": RESTIC_PASSWORD,
-                    "AWS_ACCESS_KEY_ID": AWS_ACCESS_KEY_ID,
-                    "AWS_SECRET_ACCESS_KEY": AWS_SECRET_ACCESS_KEY,
-                    "AWS_DEFAULT_REGION": AWS_DEFAULT_REGION,
+                    "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,
                 }
             )
 
@@ -308,15 +385,22 @@ class BackupManager:
 
     def send_telegram_notification(self, success: bool) -> None:
         """Send notification to Telegram"""
+        telegram_cfg = self._select_telegram()
+        if telegram_cfg is None:
+            logger.warning("No telegram notification backend configured")
+            return
+
         try:
             if success and not self.errors:
-                message = f"<b>{NOTIFICATIONS_NAME}</b>: бекап успешно завершен!"
+                message = (
+                    f"<b>{telegram_cfg.notifications_name}</b>: бекап успешно завершен!"
+                )
                 if self.successful_backups:
                     message += (
                         f"\n\nУспешные бекапы: {', '.join(self.successful_backups)}"
                     )
             else:
-                message = f"<b>{NOTIFICATIONS_NAME}</b>: бекап завершен с ошибками!"
+                message = f"<b>{telegram_cfg.notifications_name}</b>: бекап завершен с ошибками!"
 
                 if self.successful_backups:
                     message += (
@@ -329,8 +413,12 @@ class BackupManager:
                 if self.errors:
                     message += f"\n\n❌ Ошибки:\n" + "\n".join(self.errors)
 
-            url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
-            data = {"chat_id": TELEGRAM_CHAT_ID, "parse_mode": "HTML", "text": message}
+            url = f"https://api.telegram.org/bot{telegram_cfg.telegram_bot_token}/sendMessage"
+            data = {
+                "chat_id": telegram_cfg.telegram_chat_id,
+                "parse_mode": "HTML",
+                "text": message,
+            }
 
             response = requests.post(url, data=data, timeout=30)