yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-cli: Refactor the API to use the Flake object

Browse Source
This commit is contained in:
Qubasa
2025-05-07 13:12:43 +02:00
committed by Mic92
parent dd3bb314fd
commit 153da50d6f
22 changed files with 125 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkgs/clan-cli/clan_cli/machines/create.py
View File

@@ -110,7 +110,7 @@ def create_machine(opts: CreateOptions, commit: bool = True) -> None:
new_machine["deploy"] = {"targetHost": target_host}
patch_inventory_with(
clan_dir, f"machines.{machine_name}", dataclass_to_dict(new_machine)
Flake(str(clan_dir)), f"machines.{machine_name}", dataclass_to_dict(new_machine)
)
# Commit at the end in that order to avoid committing halve-baked machines

4
pkgs/clan-cli/clan_cli/machines/delete.py
View File

@@ -23,7 +23,7 @@ log = logging.getLogger(__name__)
@API.register
def delete_machine(flake: Flake, name: str) -> None:
try:
inventory.delete(str(flake.path), {f"machines.{name}"})
inventory.delete(flake, {f"machines.{name}"})
except KeyError as exc:
# louis@(2025-03-09): test infrastructure does not seem to set the
# inventory properly, but more importantly only one machine in my
@@ -35,7 +35,7 @@ def delete_machine(flake: Flake, name: str) -> None:
changed_paths: list[Path] = []
folder = specific_machine_dir(flake.path, name)
folder = specific_machine_dir(flake, name)
if folder.exists():
changed_paths.append(folder)
shutil.rmtree(folder)

22
pkgs/clan-cli/clan_cli/machines/hardware.py
View File

@@ -26,39 +26,39 @@ class HardwareConfig(Enum):
NIXOS_GENERATE_CONFIG = "nixos-generate-config"
NONE = "none"
def config_path(self, clan_dir: Path, machine_name: str) -> Path:
machine_dir = specific_machine_dir(clan_dir, machine_name)
def config_path(self, flake: Flake, machine_name: str) -> Path:
machine_dir = specific_machine_dir(flake, machine_name)
if self == HardwareConfig.NIXOS_FACTER:
return machine_dir / "facter.json"
return machine_dir / "hardware-configuration.nix"
@classmethod
def detect_type(
cls: type["HardwareConfig"], clan_dir: Path, machine_name: str
cls: type["HardwareConfig"], flake: Flake, machine_name: str
) -> "HardwareConfig":
hardware_config = HardwareConfig.NIXOS_GENERATE_CONFIG.config_path(
clan_dir, machine_name
flake, machine_name
)
if hardware_config.exists() and "throw" not in hardware_config.read_text():
return HardwareConfig.NIXOS_GENERATE_CONFIG
if HardwareConfig.NIXOS_FACTER.config_path(clan_dir, machine_name).exists():
if HardwareConfig.NIXOS_FACTER.config_path(flake, machine_name).exists():
return HardwareConfig.NIXOS_FACTER
return HardwareConfig.NONE
@API.register
def show_machine_hardware_config(clan_dir: Path, machine_name: str) -> HardwareConfig:
def show_machine_hardware_config(flake: Flake, machine_name: str) -> HardwareConfig:
"""
Show hardware information for a machine returns None if none exist.
"""
return HardwareConfig.detect_type(clan_dir, machine_name)
return HardwareConfig.detect_type(flake, machine_name)
@API.register
def show_machine_hardware_platform(clan_dir: Path, machine_name: str) -> str | None:
def show_machine_hardware_platform(flake: Flake, machine_name: str) -> str | None:
"""
Show hardware information for a machine returns None if none exist.
"""
@@ -66,7 +66,7 @@ def show_machine_hardware_platform(clan_dir: Path, machine_name: str) -> str | N
system = config["system"]
cmd = nix_eval(
[
f"{clan_dir}#clanInternals.machines.{system}.{machine_name}",
f"{flake}#clanInternals.machines.{system}.{machine_name}",
"--apply",
"machine: { inherit (machine.pkgs) system; }",
"--json",
@@ -103,7 +103,7 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
override_target_host=opts.target_host,
)
hw_file = opts.backend.config_path(opts.flake.path, opts.machine)
hw_file = opts.backend.config_path(opts.flake, opts.machine)
hw_file.parent.mkdir(parents=True, exist_ok=True)
if opts.backend == HardwareConfig.NIXOS_FACTER:
@@ -152,7 +152,7 @@ def generate_machine_hardware_info(opts: HardwareGenerateOptions) -> HardwareCon
f"machines/{opts.machine}/{hw_file.name}: update hardware configuration",
)
try:
show_machine_hardware_platform(opts.flake.path, opts.machine)
show_machine_hardware_platform(opts.flake, opts.machine)
if backup_file:
backup_file.unlink(missing_ok=True)
except ClanCmdError as e:

2
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -113,7 +113,7 @@ def install_machine(opts: InstallOptions) -> None:
str(opts.update_hardware_config.value),
str(
opts.update_hardware_config.config_path(
machine.flake.path, machine.name
machine.flake, machine.name
)
),
]

30
pkgs/clan-cli/clan_cli/machines/list.py
View File

@@ -16,12 +16,14 @@ from clan_cli.cmd import RunOpts, run
from clan_cli.completions import add_dynamic_completer, complete_tags
from clan_cli.dirs import specific_machine_dir
from clan_cli.errors import ClanError
from clan_cli.flake import Flake
from clan_cli.inventory import (
load_inventory_eval,
patch_inventory_with,
)
from clan_cli.inventory.classes import Machine as InventoryMachine
from clan_cli.machines.hardware import HardwareConfig
from clan_cli.machines.machines import Machine
from clan_cli.nix import nix_eval
from clan_cli.tags import list_nixos_machines_by_tags
@@ -29,15 +31,13 @@ log = logging.getLogger(__name__)
@API.register
def set_machine(flake_url: Path, machine_name: str, machine: InventoryMachine) -> None:
patch_inventory_with(
flake_url, f"machines.{machine_name}", dataclass_to_dict(machine)
)
def set_machine(flake: Flake, machine_name: str, machine: InventoryMachine) -> None:
patch_inventory_with(flake, f"machines.{machine_name}", dataclass_to_dict(machine))
@API.register
def list_machines(flake_url: str | Path) -> dict[str, InventoryMachine]:
inventory = load_inventory_eval(flake_url)
def list_machines(flake: Flake) -> dict[str, InventoryMachine]:
inventory = load_inventory_eval(flake)
return inventory.get("machines", {})
@@ -60,16 +60,16 @@ def extract_header(c: str) -> str:
@API.register
def get_machine_details(flake_url: Path, machine_name: str) -> MachineDetails:
inventory = load_inventory_eval(flake_url)
machine = inventory.get("machines", {}).get(machine_name)
if machine is None:
msg = f"Machine {machine_name} not found in inventory"
def get_machine_details(machine: Machine) -> MachineDetails:
inventory = load_inventory_eval(machine.flake)
machine_inv = inventory.get("machines", {}).get(machine.name)
if machine_inv is None:
msg = f"Machine {machine.name} not found in inventory"
raise ClanError(msg)
hw_config = HardwareConfig.detect_type(flake_url, machine_name)
hw_config = HardwareConfig.detect_type(machine.flake, machine.name)
machine_dir = specific_machine_dir(flake_url, machine_name)
machine_dir = specific_machine_dir(machine.flake, machine.name)
disk_schema: MachineDiskMatter | None = None
disk_path = machine_dir / "disko.nix"
if disk_path.exists():
@@ -80,7 +80,9 @@ def get_machine_details(flake_url: Path, machine_name: str) -> MachineDetails:
if data:
disk_schema = data # type: ignore
return MachineDetails(machine=machine, hw_config=hw_config, disk_schema=disk_schema)
return MachineDetails(
machine=machine_inv, hw_config=hw_config, disk_schema=disk_schema
)
def list_nixos_machines(flake_url: str | Path) -> list[str]: