Backups: use dataclass Application for app info
This commit is contained in:
@@ -10,8 +10,9 @@ import sys
|
|||||||
import subprocess
|
import subprocess
|
||||||
import logging
|
import logging
|
||||||
import pwd
|
import pwd
|
||||||
|
from dataclasses import dataclass
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import List, Tuple, Optional
|
from typing import List, Optional
|
||||||
import requests
|
import requests
|
||||||
import configparser
|
import configparser
|
||||||
|
|
||||||
@@ -47,15 +48,21 @@ BACKUP_TARGETS_FILE = "backup-targets"
|
|||||||
BACKUP_DEFAULT_DIR = "backups"
|
BACKUP_DEFAULT_DIR = "backups"
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class Application:
|
||||||
|
path: Path
|
||||||
|
owner: str
|
||||||
|
|
||||||
|
|
||||||
class BackupManager:
|
class BackupManager:
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.errors = []
|
self.errors: List[str] = []
|
||||||
self.warnings = []
|
self.warnings: List[str] = []
|
||||||
self.successful_backups = []
|
self.successful_backups: List[str] = []
|
||||||
|
|
||||||
def get_application_directories(self) -> List[Tuple[str, str]]:
|
def find_applications(self) -> List[Application]:
|
||||||
"""Get all home directories and their owners"""
|
"""Get all application directories and their owners."""
|
||||||
app_dirs = []
|
applications: List[Application] = []
|
||||||
applications_path = Path("/mnt/applications")
|
applications_path = Path("/mnt/applications")
|
||||||
source_dirs = applications_path.iterdir()
|
source_dirs = applications_path.iterdir()
|
||||||
|
|
||||||
@@ -64,14 +71,13 @@ class BackupManager:
|
|||||||
continue
|
continue
|
||||||
if app_dir.is_dir():
|
if app_dir.is_dir():
|
||||||
try:
|
try:
|
||||||
# Get the owner of the directory
|
|
||||||
stat_info = app_dir.stat()
|
stat_info = app_dir.stat()
|
||||||
owner = pwd.getpwuid(stat_info.st_uid).pw_name
|
owner = pwd.getpwuid(stat_info.st_uid).pw_name
|
||||||
app_dirs.append((str(app_dir), owner))
|
applications.append(Application(path=app_dir, owner=owner))
|
||||||
except (KeyError, OSError) as e:
|
except (KeyError, OSError) as e:
|
||||||
logger.warning(f"Could not get owner for {app_dir}: {e}")
|
logger.warning(f"Could not get owner for {app_dir}: {e}")
|
||||||
|
|
||||||
return app_dirs
|
return applications
|
||||||
|
|
||||||
def find_backup_script(self, app_dir: str) -> Optional[str]:
|
def find_backup_script(self, app_dir: str) -> Optional[str]:
|
||||||
"""Find backup script in user's home directory"""
|
"""Find backup script in user's home directory"""
|
||||||
@@ -134,7 +140,7 @@ class BackupManager:
|
|||||||
def get_backup_directories(self) -> List[str]:
|
def get_backup_directories(self) -> List[str]:
|
||||||
"""Collect backup targets according to backup-targets rules"""
|
"""Collect backup targets according to backup-targets rules"""
|
||||||
backup_dirs: List[str] = []
|
backup_dirs: List[str] = []
|
||||||
app_dirs = self.get_application_directories()
|
applications = self.find_applications()
|
||||||
|
|
||||||
def parse_targets_file(targets_file: Path) -> List[str]:
|
def parse_targets_file(targets_file: Path) -> List[str]:
|
||||||
"""Parse backup-targets file, skipping comments and empty lines."""
|
"""Parse backup-targets file, skipping comments and empty lines."""
|
||||||
@@ -151,8 +157,8 @@ class BackupManager:
|
|||||||
self.warnings.append(warning_msg)
|
self.warnings.append(warning_msg)
|
||||||
return targets
|
return targets
|
||||||
|
|
||||||
for app_dir_str, _ in app_dirs:
|
for app in applications:
|
||||||
app_dir = Path(app_dir_str)
|
app_dir = app.path
|
||||||
targets_file = app_dir / BACKUP_TARGETS_FILE
|
targets_file = app_dir / BACKUP_TARGETS_FILE
|
||||||
resolved_targets: List[Path] = []
|
resolved_targets: List[Path] = []
|
||||||
|
|
||||||
@@ -317,11 +323,13 @@ class BackupManager:
|
|||||||
logger.info("Starting backup process")
|
logger.info("Starting backup process")
|
||||||
|
|
||||||
# Get all home directories
|
# Get all home directories
|
||||||
app_dirs = self.get_application_directories()
|
applications = self.find_applications()
|
||||||
logger.info(f"Found {len(app_dirs)} application directories")
|
logger.info(f"Found {len(applications)} application directories")
|
||||||
|
|
||||||
# Process each user's backup
|
# Process each user's backup
|
||||||
for app_dir, username in app_dirs:
|
for app in applications:
|
||||||
|
app_dir = str(app.path)
|
||||||
|
username = app.owner
|
||||||
logger.info(f"Processing backup for app: {app_dir} (user {username})")
|
logger.info(f"Processing backup for app: {app_dir} (user {username})")
|
||||||
|
|
||||||
# Find backup script
|
# Find backup script
|
||||||
|
|||||||
Reference in New Issue
Block a user