From 21b0c00c05367347566ef70cf9ea4ba178a3c96d Mon Sep 17 00:00:00 2001 From: Qubasa Date: Tue, 24 Jun 2025 11:32:42 +0200 Subject: [PATCH] clan-cli: Move list.py to clan_lib/machines --- pkgs/clan-cli/clan_cli/machines/list.py | 101 +--------------------- pkgs/clan-cli/clan_lib/machines/list.py | 106 ++++++++++++++++++++++++ 2 files changed, 107 insertions(+), 100 deletions(-) create mode 100644 pkgs/clan-cli/clan_lib/machines/list.py diff --git a/pkgs/clan-cli/clan_cli/machines/list.py b/pkgs/clan-cli/clan_cli/machines/list.py index 582dd6f0d..84ed970cb 100644 --- a/pkgs/clan-cli/clan_cli/machines/list.py +++ b/pkgs/clan-cli/clan_cli/machines/list.py @@ -1,113 +1,14 @@ import argparse import logging -import re -from dataclasses import dataclass -from clan_lib.api import API -from clan_lib.api.disk import MachineDiskMatter -from clan_lib.api.modules import parse_frontmatter -from clan_lib.dirs import specific_machine_dir -from clan_lib.errors import ClanError from clan_lib.flake import Flake -from clan_lib.machines.actions import get_machine, list_machines -from clan_lib.machines.machines import Machine -from clan_lib.nix_models.clan import InventoryMachine +from clan_lib.machines.list import list_full_machines, query_machines_by_tags from clan_cli.completions import add_dynamic_completer, complete_tags -from clan_cli.machines.hardware import HardwareConfig log = logging.getLogger(__name__) -def list_full_machines( - flake: Flake, nix_options: list[str] | None = None -) -> dict[str, Machine]: - """ - Like `list_machines`, but returns a full 'machine' instance for each machine. - """ - machines = list_machines(flake) - - res: dict[str, Machine] = {} - - if nix_options is None: - nix_options = [] - - for inv_machine in machines.values(): - name = inv_machine.get("name") - # Technically, this should not happen, but we are defensive here. - if name is None: - msg = "InternalError: Machine name is required. But got a machine without a name." - raise ClanError(msg) - - machine = Machine( - name=name, - flake=flake, - nix_options=nix_options, - ) - res[machine.name] = machine - - return res - - -def query_machines_by_tags(flake: Flake, tags: list[str]) -> dict[str, Machine]: - """ - 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_full_machines(flake) - - filtered_machines = {} - for machine in machines.values(): - inv_machine = get_machine(machine.flake, machine.name) - machine_tags = inv_machine.get("tags", []) - if all(tag in machine_tags for tag in tags): - filtered_machines[machine.name] = machine - - return filtered_machines - - -@dataclass -class MachineDetails: - machine: InventoryMachine - hw_config: HardwareConfig | None = None - disk_schema: MachineDiskMatter | None = None - - -def extract_header(c: str) -> str: - header_lines = [] - for line in c.splitlines(): - match = re.match(r"^\s*#(.*)", line) - if match: - header_lines.append(match.group(1).strip()) - else: - break # Stop once the header ends - return "\n".join(header_lines) - - -@API.register -def get_machine_details(machine: Machine) -> MachineDetails: - machine_inv = get_machine(machine.flake, machine.name) - hw_config = HardwareConfig.detect_type(machine) - - machine_dir = specific_machine_dir(machine) - disk_schema: MachineDiskMatter | None = None - disk_path = machine_dir / "disko.nix" - if disk_path.exists(): - with disk_path.open() as f: - content = f.read() - header = extract_header(content) - data, _rest = parse_frontmatter(header) - if data: - disk_schema = data # type: ignore - - return MachineDetails( - machine=machine_inv, - hw_config=hw_config, - disk_schema=disk_schema, - ) - - def list_command(args: argparse.Namespace) -> None: flake: Flake = args.flake diff --git a/pkgs/clan-cli/clan_lib/machines/list.py b/pkgs/clan-cli/clan_lib/machines/list.py new file mode 100644 index 000000000..6232e3017 --- /dev/null +++ b/pkgs/clan-cli/clan_lib/machines/list.py @@ -0,0 +1,106 @@ +import logging +import re +from dataclasses import dataclass + +from clan_cli.machines.hardware import HardwareConfig + +from clan_lib.api import API +from clan_lib.api.disk import MachineDiskMatter +from clan_lib.api.modules import parse_frontmatter +from clan_lib.dirs import specific_machine_dir +from clan_lib.errors import ClanError +from clan_lib.flake import Flake +from clan_lib.machines.actions import get_machine, list_machines +from clan_lib.machines.machines import Machine +from clan_lib.nix_models.clan import InventoryMachine + +log = logging.getLogger(__name__) + + +def list_full_machines( + flake: Flake, nix_options: list[str] | None = None +) -> dict[str, Machine]: + """ + Like `list_machines`, but returns a full 'machine' instance for each machine. + """ + machines = list_machines(flake) + + res: dict[str, Machine] = {} + + if nix_options is None: + nix_options = [] + + for inv_machine in machines.values(): + name = inv_machine.get("name") + # Technically, this should not happen, but we are defensive here. + if name is None: + msg = "InternalError: Machine name is required. But got a machine without a name." + raise ClanError(msg) + + machine = Machine( + name=name, + flake=flake, + nix_options=nix_options, + ) + res[machine.name] = machine + + return res + + +def query_machines_by_tags(flake: Flake, tags: list[str]) -> dict[str, Machine]: + """ + 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_full_machines(flake) + + filtered_machines = {} + for machine in machines.values(): + inv_machine = get_machine(machine.flake, machine.name) + machine_tags = inv_machine.get("tags", []) + if all(tag in machine_tags for tag in tags): + filtered_machines[machine.name] = machine + + return filtered_machines + + +@dataclass +class MachineDetails: + machine: InventoryMachine + hw_config: HardwareConfig | None = None + disk_schema: MachineDiskMatter | None = None + + +def extract_header(c: str) -> str: + header_lines = [] + for line in c.splitlines(): + match = re.match(r"^\s*#(.*)", line) + if match: + header_lines.append(match.group(1).strip()) + else: + break # Stop once the header ends + return "\n".join(header_lines) + + +@API.register +def get_machine_details(machine: Machine) -> MachineDetails: + machine_inv = get_machine(machine.flake, machine.name) + hw_config = HardwareConfig.detect_type(machine) + + machine_dir = specific_machine_dir(machine) + disk_schema: MachineDiskMatter | None = None + disk_path = machine_dir / "disko.nix" + if disk_path.exists(): + with disk_path.open() as f: + content = f.read() + header = extract_header(content) + data, _rest = parse_frontmatter(header) + if data: + disk_schema = data # type: ignore + + return MachineDetails( + machine=machine_inv, + hw_config=hw_config, + disk_schema=disk_schema, + )