backup: add application finder class

This commit is contained in:
2026-04-12 17:42:43 +03:00
parent b7a18f1296
commit 0aa34efd00
+33 -30
View File
@@ -43,7 +43,6 @@ logger = logging.getLogger(__name__)
@dataclass
class Config:
host_name: str
roots: List[Path]
@dataclass
@@ -188,21 +187,9 @@ class AppriseNotifier(Notifier):
)
class BackupManager:
def __init__(
self,
config: Config,
roots: List[Path],
storages: List[Storage],
notifiers: List[Notifier],
):
self.errors: List[str] = []
self.warnings: List[str] = []
self.successful_backups: List[str] = []
self.config = config
self.roots: List[Path] = roots
self.storages = storages
self.notifiers = notifiers
class ApplicationFinder:
def __init__(self, roots: List[Path]):
self.roots = roots
def find_applications(self) -> List[Application]:
"""Get all application directories and their owners."""
@@ -222,6 +209,21 @@ class BackupManager:
return applications
class BackupManager:
def __init__(
self,
config: Config,
storages: List[Storage],
notifiers: List[Notifier],
):
self.errors: List[str] = []
self.warnings: List[str] = []
self.successful_backups: List[str] = []
self.config = config
self.storages = storages
self.notifiers = notifiers
def find_backup_script(self, app_dir: str) -> Optional[str]:
"""Find backup script in user's home directory"""
possible_scripts = [
@@ -280,10 +282,9 @@ class BackupManager:
self.errors.append(f"App {username}: {error_msg}")
return False
def get_backup_directories(self) -> List[str]:
def get_backup_directories(self, applications: List[Application]) -> List[str]:
"""Collect backup targets according to backup-targets rules"""
backup_dirs: List[str] = []
applications = self.find_applications()
def parse_targets_file(targets_file: Path) -> List[str]:
"""Parse backup-targets file, skipping comments and empty lines."""
@@ -369,12 +370,9 @@ class BackupManager:
except Exception as e:
logger.error(f"Failed to send notification: {str(e)}")
def run_backup_process(self) -> bool:
def run_backup_process(self, applications: List[Application]) -> bool:
"""Main backup process"""
logger.info("Starting backup process")
# Get all home directories
applications = self.find_applications()
logger.info(f"Found {len(applications)} application directories")
# Process each user's backup
@@ -397,7 +395,7 @@ class BackupManager:
self.run_app_backup(backup_script, app_dir, username)
# Get backup directories
backup_dirs = self.get_backup_directories()
backup_dirs = self.get_backup_directories(applications)
logger.info(f"Found backup directories: {backup_dirs}")
overall_success = True
@@ -426,7 +424,9 @@ class BackupManager:
return True
def initialize(config_path: Path) -> BackupManager:
def initialize(
config_path: Path,
) -> tuple[ApplicationFinder, BackupManager]:
try:
with config_path.open("rb") as config_file:
raw_config = tomllib.load(config_file)
@@ -463,17 +463,20 @@ def initialize(config_path: Path) -> BackupManager:
if not notifiers:
raise ValueError("At least one notification backend must be configured")
config = Config(host_name=host_name, roots=roots)
return BackupManager(
config=config, roots=roots, storages=storages, notifiers=notifiers
config = Config(host_name=host_name)
app_finder = ApplicationFinder(roots)
backup_manager = BackupManager(
config=config, storages=storages, notifiers=notifiers
)
return app_finder, backup_manager
def main() -> None:
try:
backup_manager = initialize(CONFIG_PATH)
success = backup_manager.run_backup_process()
app_finder, backup_manager = initialize(CONFIG_PATH)
applications = app_finder.find_applications()
success = backup_manager.run_backup_process(applications)
if not success:
sys.exit(1)
except KeyboardInterrupt: