Merge pull request 'pkgs/cli: Add tagging support to machines list' (#2388) from kenji/clan-core:kenji-cli/2374/machine-list-tags into main

pkgs/clan-cli/clan_cli/machines/cli.py

@@ -57,6 +57,13 @@ Examples:
 
   $ clan machines list
   Lists all the machines and their descriptions.
+
+  $ clan machines list --tags [TAGS..]
+  Lists all the machines that have the specified tags associated through the inventory.
+  If multiple tags are specified machines are matched against both tags.
+
+  $ clan machines list --tags vm
+  Lists all machines that are associated with the "vm" tag through the inventory.
         """
         ),
         formatter_class=argparse.RawTextHelpFormatter,

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

@@ -7,9 +7,11 @@ from typing import Literal
 
 from clan_cli.api import API
 from clan_cli.cmd import run_no_stdout
+from clan_cli.completions import add_dynamic_completer, complete_tags
 from clan_cli.errors import ClanCmdError, ClanError
 from clan_cli.inventory import Machine, load_inventory_eval, set_inventory
 from clan_cli.nix import nix_eval, nix_shell
+from clan_cli.tags import list_nixos_machines_by_tags
 
 log = logging.getLogger(__name__)
 
@@ -132,9 +134,19 @@ def check_machine_online(
 
 def list_command(args: argparse.Namespace) -> None:
     flake_path = args.flake.path
+    if args.tags:
+        list_nixos_machines_by_tags(flake_path, args.tags)
+        return
     for name in list_nixos_machines(flake_path):
         print(name)
 
 
 def register_list_parser(parser: argparse.ArgumentParser) -> None:
+    tag_parser = parser.add_argument(
+        "--tags",
+        nargs="+",
+        default=[],
+        help="Tags that machines should be queried for. Multiple tags will intersect.",
+    )
+    add_dynamic_completer(tag_parser, complete_tags)
     parser.set_defaults(func=list_command)

pkgs/clan-cli/clan_cli/tags.py

@@ -0,0 +1,58 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from clan_cli.cmd import run_no_stdout
+from clan_cli.errors import ClanError
+from clan_cli.nix import nix_eval
+
+
+def list_tagged_machines(flake_url: str | Path) -> dict[str, Any]:
+    """
+    Query machines from the inventory with their meta information intact.
+    The meta information includes tags.
+    """
+    cmd = nix_eval(
+        [
+            f"{flake_url}#clanInternals.inventory.machines",
+            "--json",
+        ]
+    )
+    proc = run_no_stdout(cmd)
+
+    try:
+        res = proc.stdout.strip()
+        data = json.loads(res)
+    except json.JSONDecodeError as e:
+        msg = f"Error decoding tagged inventory machines from flake: {e}"
+        raise ClanError(msg) from e
+    else:
+        return data
+
+
+def query_machines_by_tags(
+    flake_path: str | Path, tags: list[str] | None = None
+) -> list[str]:
+    """
+    Query machines by their respective tags, if multiple tags are specified
+    then only machines that have those respective tags specified will be listed.
+    It is an intersection of the tags and machines.
+    """
+    machines = list_tagged_machines(flake_path)
+
+    if not tags:
+        return list(machines.keys())
+
+    filtered_machines = []
+    for machine_id, machine_values in machines.items():
+        if all(tag in machine_values["tags"] for tag in tags):
+            filtered_machines.append(machine_id)
+
+    return filtered_machines
+
+
+def list_nixos_machines_by_tags(
+    flake_path: str | Path, tags: list[str] | None = None
+) -> None:
+    for name in query_machines_by_tags(flake_path, tags):
+        print(name)