av/pet-project-server
1
0
Fork 0
Code Pull Requests Actions Activity

Backup: support for multiple storages
Some checks failed
Linting / YAML Lint (push) Failing after 9s
Details
Linting / Ansible Lint (push) Successful in 19s
Details

Browse Source
This commit is contained in:
Anton Vakhrushev
2025-12-20 21:19:06 +03:00
parent 0e96b5030d
commit 2655869814
1 changed files with 56 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
files/backups/backup-all.py
View File

@@ -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():