pkgs/cli: Add tagging support to machines list

Add the `--tags` flag to `clan machines list`
This now supports the machine tagging system from the inventory.

Multiple tags are the intersection of the tags of a specific machine.

Example two machines with overlapping tags:
```
server: ["intel"]
laptop: ["intel", "graphical"]
```

- `clan machines list --tags intel` will output:

```
server
laptop
```

- `clan machines list --tags intel graphical` will output:

```
laptop
```

- `clan machines list --tags graphical` will output:

```
laptop
```
This commit is contained in:
a-kenji
2024-11-12 14:57:06 +01:00
parent 642cb434c2
commit 8a6239e08d
2 changed files with 68 additions and 0 deletions

View File

@@ -10,6 +10,7 @@ from clan_cli.cmd import run_no_stdout
from clan_cli.errors import ClanCmdError, ClanError from clan_cli.errors import ClanCmdError, ClanError
from clan_cli.inventory import Machine, load_inventory_eval, set_inventory from clan_cli.inventory import Machine, load_inventory_eval, set_inventory
from clan_cli.nix import nix_eval, nix_shell from clan_cli.nix import nix_eval, nix_shell
from clan_cli.tags import list_nixos_machines_by_tags
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -132,9 +133,18 @@ def check_machine_online(
def list_command(args: argparse.Namespace) -> None: def list_command(args: argparse.Namespace) -> None:
flake_path = args.flake.path 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): for name in list_nixos_machines(flake_path):
print(name) print(name)
def register_list_parser(parser: argparse.ArgumentParser) -> None: 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.",
)
parser.set_defaults(func=list_command) parser.set_defaults(func=list_command)

View File

@@ -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)