Merge pull request 'clan: add dynamic completions to clan backups' (#1565) from kenji/clan-core:clan/backups/dynamic-completions into main

Reviewed-on: clan/clan-core#1565

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

@@ -2,7 +2,11 @@ import argparse
 import json
 import logging
 
-from ..completions import add_dynamic_completer, complete_machines
+from ..completions import (
+    add_dynamic_completer,
+    complete_backup_providers_for_machine,
+    complete_machines,
+)
 from ..errors import ClanError
 from ..machines.machines import Machine
 
@@ -46,5 +50,8 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
     )
     add_dynamic_completer(machines_parser, complete_machines)
 
-    parser.add_argument("--provider", type=str, help="backup provider to use")
+    provider_action = parser.add_argument(
+        "--provider", type=str, help="backup provider to use"
+    )
+    add_dynamic_completer(provider_action, complete_backup_providers_for_machine)
     parser.set_defaults(func=create_command)

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

@@ -3,7 +3,11 @@ import json
 import subprocess
 from dataclasses import dataclass
 
-from ..completions import add_dynamic_completer, complete_machines
+from ..completions import (
+    add_dynamic_completer,
+    complete_backup_providers_for_machine,
+    complete_machines,
+)
 from ..errors import ClanError
 from ..machines.machines import Machine
 
@@ -62,5 +66,8 @@ def register_list_parser(parser: argparse.ArgumentParser) -> None:
         "machine", type=str, help="machine in the flake to show backups of"
     )
     add_dynamic_completer(machines_parser, complete_machines)
-    parser.add_argument("--provider", type=str, help="backup provider to filter by")
+    provider_action = parser.add_argument(
+        "--provider", type=str, help="backup provider to filter by"
+    )
+    add_dynamic_completer(provider_action, complete_backup_providers_for_machine)
     parser.set_defaults(func=list_command)

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

@@ -2,6 +2,11 @@ import argparse
 import json
 import subprocess
 
+from ..completions import (
+    add_dynamic_completer,
+    complete_backup_providers_for_machine,
+    complete_machines,
+)
 from ..errors import ClanError
 from ..machines.machines import Machine
 
@@ -74,10 +79,14 @@ def restore_command(args: argparse.Namespace) -> None:
 
 
 def register_restore_parser(parser: argparse.ArgumentParser) -> None:
-    parser.add_argument(
+    machine_action = parser.add_argument(
         "machine", type=str, help="machine in the flake to create backups of"
     )
-    parser.add_argument("provider", type=str, help="backup provider to use")
+    add_dynamic_completer(machine_action, complete_machines)
+    provider_action = parser.add_argument(
+        "provider", type=str, help="backup provider to use"
+    )
+    add_dynamic_completer(provider_action, complete_backup_providers_for_machine)
     parser.add_argument("name", type=str, help="Name of the backup to restore")
     parser.add_argument("--service", type=str, help="name of the service to restore")
     parser.set_defaults(func=restore_command)

pkgs/clan-cli/clan_cli/completions.py

@@ -118,6 +118,49 @@ def complete_services_for_machine(
     return services_dict
 
 
+def complete_backup_providers_for_machine(
+    prefix: str, parsed_args: argparse.Namespace, **kwargs: Any
+) -> Iterable[str]:
+    """
+    Provides completion functionality for machine backup providers.
+    """
+    services: list[str] = []
+    # TODO: consolidate, if multiple machines are used
+    machines: list[str] = parsed_args.machines
+
+    def run_cmd() -> None:
+        try:
+            if (clan_dir_result := clan_dir(None)) is not None:
+                flake = clan_dir_result
+            else:
+                flake = "."
+            services_result = json.loads(
+                run(
+                    nix_eval(
+                        flags=[
+                            f"{flake}#nixosConfigurations.{machines[0]}.config.clanCore.backups.providers",
+                            "--apply",
+                            "builtins.attrNames",
+                        ],
+                    ),
+                ).stdout.strip()
+            )
+
+            services.extend(services_result)
+        except subprocess.CalledProcessError:
+            pass
+
+    thread = threading.Thread(target=run_cmd)
+    thread.start()
+    thread.join(timeout=COMPLETION_TIMEOUT)
+
+    if thread.is_alive():
+        return iter([])
+
+    services_dict = {name: "service" for name in services}
+    return services_dict
+
+
 def complete_secrets(
     prefix: str, parsed_args: argparse.Namespace, **kwargs: Any
 ) -> Iterable[str]: