helpers file optimizations

src/helpers.py

@@ -5,6 +5,11 @@ from datetime import datetime
 from subprocess import run
 import os
 import typer
+from typing import Optional
+
+# Constants
+JSON_TARGETS = Path(__file__).parent / "targets.json"
+EXCLUDES_PATH = Path(__file__).parent / "excludes.txt"
 
 # Default exclude patterns for backups
 default_excludes = [
@@ -48,102 +53,191 @@ default_excludes = [
 
 
 def ask_repokey() -> None:
+    """
+    Prompt user for Borg repository key and set it as environment variable.
+    """
     repokey = typer.prompt("Enter repository key: ", hide_input=True)
     os.environ["BORG_PASSPHRASE"] = repokey
 
 
 def get_targets() -> list[BackupItem]:
-    JSON_TARGETS = Path(__file__).parent / "targets.json"
+    """
-    with open(JSON_TARGETS, "r") as f:
+    Load and parse backup targets from the JSON configuration file.
-        targets = json.load(f)
+    
-    return [BackupItem(**target) for target in targets]
+    Returns:
+        List of BackupItem objects representing the targets.
+        Empty list if the file doesn't exist or contains invalid data.
+    """
+    try:
+        with open(JSON_TARGETS, "r") as f:
+            targets = json.load(f)
+        return [BackupItem(**target) for target in targets]
+    except (json.JSONDecodeError, FileNotFoundError) as e:
+        typer.echo(f"Error loading targets: {e}")
+        return []
 
 
-def add_target(target: BackupItem) -> None:
+def add_target(target: BackupItem) -> bool:
-    # Get current targets
+    """
-    targets = get_targets()
+    Add a new backup target to the configuration.
     
-    # Add new target
+    Args:
-    targets.append(target)
+        target: BackupItem object to add
         
-    # Save targets
+    Returns:
-    write_targets(targets)
+        bool: True if successful, False otherwise
+    """
+    try:
+        # Get current targets
+        targets = get_targets()
+
+        # Add new target
+        targets.append(target)
+
+        # Save targets
+        write_targets(targets)
+        return True
+    except Exception as e:
+        typer.echo(f"Error adding target: {e}")
+        return False
 
 
-def write_targets(targets: list[BackupItem]) -> None:
+def write_targets(targets: list[BackupItem]) -> bool:
-    """Save targets to JSON file"""
+    """
-    JSON_TARGETS = Path(__file__).parent / "targets.json"
+    Save targets to JSON file.
-    with open(JSON_TARGETS, "w") as f:
+    
-        json.dump(
+    Args:
-            [target.json_serializable for target in targets],
+        targets: List of BackupItem objects to save
-            f,
+        
-            ensure_ascii=False,
+    Returns:
-            indent=2,
+        bool: True if successful, False otherwise
-        )
+    """
+    try:
+        with open(JSON_TARGETS, "w") as f:
+            json.dump(
+                [target.json_serializable for target in targets],
+                f,
+                ensure_ascii=False,
+                indent=2,
+            )
+        return True
+    except Exception as e:
+        typer.echo(f"Error writing targets: {e}")
+        return False
 
 
-def set_targets(targets: list[BackupItem]) -> None:
+# The set_targets function has been removed as it was redundant with write_targets
-    JSON_TARGETS = Path(__file__).parent / "targets.json"
-    with open(JSON_TARGETS, "w") as f:
-        json.dump(
-            [target.json_serializable for target in targets],
-            f,
-            ensure_ascii=False,
-            indent=2,
-        )
 
 
-def prune_backups(targets: list[BackupItem]) -> None:
+def prune_backups(targets: list[BackupItem], daily: int = 1, weekly: int = 2, 
+                 monthly: int = 3, yearly: int = 1) -> list[dict]:
+    """
+    Prune old backups according to retention policy and compact the repository.
+    
+    Args:
+        targets: List of backup targets to prune
+        daily: Number of daily backups to keep
+        weekly: Number of weekly backups to keep
+        monthly: Number of monthly backups to keep
+        yearly: Number of yearly backups to keep
+        
+    Returns:
+        List of dictionaries containing results for each target
+    """
+    results = []
+    
     for target in targets:
         if not target.enabled:
             continue
+            
+        result = {"target": target.path, "success": False, "message": ""}
+        
+        try:
+            cmd = [
+                "borg",
+                "prune",
+                "--keep-daily", str(daily),
+                "--keep-weekly", str(weekly),
+                "--keep-monthly", str(monthly),
+                "--keep-yearly", str(yearly),
+                target.target,
+            ]
+            print("Pruning target:", target.path)
+            prune_result = run(cmd, capture_output=True, text=True)
+            
+            if prune_result.returncode != 0:
+                result["message"] = f"Prune failed: {prune_result.stderr}"
+                results.append(result)
+                continue
+                
+            print("Pruned target:", target.path)
+            print("Compacting cache")
+            compact_result = run(["borg", "compact", target.target], capture_output=True, text=True)
+            
+            if compact_result.returncode != 0:
+                result["message"] = f"Compact failed: {compact_result.stderr}"
+                results.append(result)
+                continue
+                
+            print("Compacted cache")
+            result["success"] = True
+            result["message"] = "Pruned and compacted successfully"
+            
+        except Exception as e:
+            result["message"] = f"Error during pruning: {str(e)}"
+            
+        results.append(result)
+        
+    return results
+
+
+def run_backup(target: BackupItem) -> Optional[ResultInfo]:
+    """
+    Execute backup for the specified target using borg.
+    
+    Args:
+        target: The backup target configuration
+        
+    Returns:
+        ResultInfo object with backup statistics or None if backup failed
+        
+    Raises:
+        Exception: If the backup process fails
+    """
+    date_str = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+    
+    try:
+        # Build borg command with base options
         cmd = [
             "borg",
-            "prune",
+            "create",
-            "--keep-daily",
+            "-C",
-            "1",
+            f"{target.compression.value},{target.comp_ratio}",
-            "--keep-weekly",
+            "--exclude-from",
-            "2",
+            EXCLUDES_PATH,
-            "--keep-monthly",
+            "--stats",
-            "3",
+            "--json",
-            "--keep-yearly",
+            f"{target.target}::{date_str}",
-            "1",
+            target.path,
-            target.target,
         ]
-        print("Pruning target:", target.path)
-        run(cmd, capture_output=True)
-        print("Pruned target:", target.path)
-        print("Compacting cache")
-        run(["borg", "compact", target.target], capture_output=True)
-        print("Compacted cache")
 
+        # Add exclude option only if there are excludes defined
+        if target.excludes:  # This is true if the list is not empty
+            cmd.insert(3, "--exclude")
+            cmd.insert(4, ",".join(target.excludes))
             
-def run_backup(target: BackupItem) -> None:
+        print("Running target:", target.path)
-    date_str = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+        result = run(cmd, capture_output=True)
-    # Build borg command with base options
-    cmd = [
-        "borg",
-        "create",
-        "-C",
-        f"{target.compression.value},{target.comp_ratio}",
-        "--exclude-from",
-        "/home/esa/bin/backupper/src/excludes.txt",
-        "--stats",
-        "--json",
-        f"{target.target}::{date_str}",
-        target.path,
-    ]
         
-    # Add exclude option only if there are excludes defined
+        if result.returncode != 0:
-    if target.excludes:  # This is true if the list is not empty
+            print("Result:", result.stderr.decode("utf-8"))
-        cmd.insert(3, "--exclude")
+            raise Exception(result.stderr.decode("utf-8"))
-        cmd.insert(4, ",".join(target.excludes))
-    print("Running target:", target.path)
-    result = run(cmd, capture_output=True)
-    if result.returncode != 0:
-        print("Result:", result.stderr.decode("utf-8"))
-        raise Exception(result.stderr.decode("utf-8"))
-    result_info = ResultInfo(**json.loads(result.stdout.decode("utf-8")))
-    prune_backups([target])
             
-    return result_info
+        result_info = ResultInfo(**json.loads(result.stdout.decode("utf-8")))
+        prune_backups([target])
+
+        return result_info
+        
+    except Exception as e:
+        print(f"Backup failed: {str(e)}")
+        return None