backups: support services for restore

pkgs/clan-cli/clan_cli/backups/create.py

@@ -1,6 +1,5 @@
 import argparse
 import json
-import subprocess
 
 from ..errors import ClanError
 from ..machines.machines import Machine
@@ -12,8 +11,8 @@ def create_backup(machine: Machine, provider: str | None = None) -> None:
     )
     if provider is None:
         for provider in backup_scripts["providers"]:
-            proc = subprocess.run(
-                ["bash", "-c", backup_scripts["providers"][provider]["start"]],
+            proc = machine.host.run(
+                ["bash", "-c", backup_scripts["providers"][provider]["create"]],
             )
             if proc.returncode != 0:
                 raise ClanError("failed to start backup")
@@ -22,8 +21,8 @@ def create_backup(machine: Machine, provider: str | None = None) -> None:
     else:
         if provider not in backup_scripts["providers"]:
             raise ClanError(f"provider {provider} not found")
-        proc = subprocess.run(
-            ["bash", "-c", backup_scripts["providers"][provider]["start"]],
+        proc = machine.host.run(
+            ["bash", "-c", backup_scripts["providers"][provider]["create"]],
         )
         if proc.returncode != 0:
             raise ClanError("failed to start backup")

pkgs/clan-cli/clan_cli/backups/list.py

@@ -1,47 +1,68 @@
 import argparse
 import json
 import subprocess
-from typing import Any
+from dataclasses import dataclass
 
 from ..errors import ClanError
 from ..machines.machines import Machine
 
 
-def list_backups(machine: Machine, provider: str | None = None) -> list[dict[str, Any]]:
-    backup_scripts = json.loads(
+@dataclass
+class Backup:
+    archive_id: str
+    date: str
+    provider: str
+    remote_path: str
+    job_name: str
+
+
+def list_provider(machine: Machine, provider: str) -> list[Backup]:
+    results = []
+    backup_metadata = json.loads(
+        machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.backups")
+    )
+    proc = machine.host.run(
+        ["bash", "-c", backup_metadata["providers"][provider]["list"]],
+        stdout=subprocess.PIPE,
+        check=False,
+    )
+    if proc.returncode != 0:
+        # TODO this should be a warning, only raise exception if no providers succeed
+        ClanError(f"failed to list backups for provider {provider}")
+    else:
+        parsed_json = json.loads(proc.stdout)
+        # TODO move borg specific code to borgbackup.nix
+        for archive in parsed_json["archives"]:
+            backup = Backup(
+                archive_id=archive["archive"],
+                date=archive["time"],
+                provider=provider,
+                remote_path=parsed_json["repository"]["location"],
+                job_name=parsed_json["job-name"],
+            )
+            results.append(backup)
+    return results
+
+
+def list_backups(machine: Machine, provider: str | None = None) -> list[Backup]:
+    backup_metadata = json.loads(
         machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.backups")
     )
     results = []
     if provider is None:
-        for provider in backup_scripts["providers"]:
-            proc = subprocess.run(
-                ["bash", "-c", backup_scripts["providers"][provider]["list"]],
-                stdout=subprocess.PIPE,
-            )
-            if proc.returncode != 0:
-                # TODO this should be a warning, only raise exception if no providers succeed
-                raise ClanError("failed to list backups")
-            else:
-                results.append(proc.stdout)
-    else:
-        if provider not in backup_scripts["providers"]:
-            raise ClanError(f"provider {provider} not found")
-        proc = subprocess.run(
-            ["bash", "-c", backup_scripts["providers"][provider]["list"]],
-            stdout=subprocess.PIPE,
-        )
-        if proc.returncode != 0:
-            raise ClanError("failed to list backup")
-        else:
-            results.append(proc.stdout)
+        for _provider in backup_metadata["providers"]:
+            results += list_provider(machine, _provider)
 
-    return list(map(json.loads, results))
+    else:
+        results += list_provider(machine, provider)
+
+    return results
 
 
 def list_command(args: argparse.Namespace) -> None:
     machine = Machine(name=args.machine, flake_dir=args.flake)
-    backups_data = list_backups(machine=machine, provider=args.provider)
-    print(json.dumps(list(backups_data)))
+    backups = list_backups(machine=machine, provider=args.provider)
+    print(backups)
 
 
 def register_list_parser(parser: argparse.ArgumentParser) -> None:

pkgs/clan-cli/clan_cli/backups/restore.py

@@ -2,64 +2,100 @@ import argparse
 import json
 import os
 import subprocess
-from typing import Any
 
 from ..errors import ClanError
 from ..machines.machines import Machine
-from .list import list_backups
+from .list import Backup, list_backups
 
 
-def restore_backup(
-    backup_data: list[dict[str, Any]],
-    machine: Machine,
-    provider: str,
-    archive_id: str,
-    service: str | None = None,
+def restore_service(
+    machine: Machine, backup: Backup, provider: str, service: str
 ) -> None:
-    backup_scripts = json.loads(
+    backup_metadata = json.loads(
         machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.backups")
     )
     backup_folders = json.loads(
         machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.state")
     )
-    if service is None:
-        for backup in backup_data:
-            for archive in backup["archives"]:
-                if archive["archive"] == archive_id:
-                    env = os.environ.copy()
-                    env["ARCHIVE_ID"] = archive_id
-                    env["LOCATION"] = backup["repository"]["location"]
-                    env["JOB"] = backup["job-name"]
-                    proc = subprocess.run(
-                        [
-                            "bash",
-                            "-c",
-                            backup_scripts["providers"][provider]["restore"],
-                        ],
-                        stdout=subprocess.PIPE,
-                        env=env,
-                    )
-                    if proc.returncode != 0:
-                        # TODO this should be a warning, only raise exception if no providers succeed
-                        raise ClanError("failed to restore backup")
-    else:
-        print(
-            "would restore backup",
-            machine,
-            provider,
-            archive_id,
-            "of service:",
-            service,
+    folders = backup_folders[service]["folders"]
+    env = os.environ.copy()
+    env["ARCHIVE_ID"] = backup.archive_id
+    env["LOCATION"] = backup.remote_path
+    env["JOB"] = backup.job_name
+    env["FOLDERS"] = ":".join(folders)
+
+    proc = machine.host.run(
+        [
+            "bash",
+            "-c",
+            backup_folders[service]["preRestoreScript"],
+        ],
+        stdout=subprocess.PIPE,
+        extra_env=env,
+    )
+    if proc.returncode != 0:
+        raise ClanError(
+            f"failed to run preRestoreScript: {backup_folders[service]['preRestoreScript']}, error was: {proc.stdout}"
         )
-        print(backup_folders)
+
+    proc = machine.host.run(
+        [
+            "bash",
+            "-c",
+            backup_metadata["providers"][provider]["restore"],
+        ],
+        stdout=subprocess.PIPE,
+        extra_env=env,
+    )
+    if proc.returncode != 0:
+        raise ClanError(
+            f"failed to restore backup: {backup_metadata['providers'][provider]['restore']}"
+        )
+
+    proc = machine.host.run(
+        [
+            "bash",
+            "-c",
+            backup_folders[service]["postRestoreScript"],
+        ],
+        stdout=subprocess.PIPE,
+        extra_env=env,
+    )
+    if proc.returncode != 0:
+        raise ClanError(
+            f"failed to run postRestoreScript: {backup_folders[service]['postRestoreScript']}, error was: {proc.stdout}"
+        )
+
+
+def restore_backup(
+    machine: Machine,
+    backups: list[Backup],
+    provider: str,
+    archive_id: str,
+    service: str | None = None,
+) -> None:
+    if service is None:
+        for backup in backups:
+            if backup.archive_id == archive_id:
+                backup_folders = json.loads(
+                    machine.eval_nix(
+                        f"nixosConfigurations.{machine.name}.config.clanCore.state"
+                    )
+                )
+                for _service in backup_folders:
+                    restore_service(machine, backup, provider, _service)
+    else:
+        for backup in backups:
+            if backup.archive_id == archive_id:
+                restore_service(machine, backup, provider, service)
 
 
 def restore_command(args: argparse.Namespace) -> None:
     machine = Machine(name=args.machine, flake_dir=args.flake)
-    backup_data = list_backups(machine=machine, provider=args.provider)
+    backups = list_backups(machine=machine, provider=args.provider)
     restore_backup(
-        backup_data=backup_data,
         machine=machine,
+        backups=backups,
         provider=args.provider,
         archive_id=args.archive_id,
         service=args.service,