inventory.{cli,api}: use only dictionaries

This commit is contained in:
Johannes Kirschbauer
2024-12-06 18:50:49 +01:00
parent df43639e9b
commit 69b7f6be5b
11 changed files with 95 additions and 50 deletions

View File

@@ -16,7 +16,6 @@ from clan_cli.git import commit_file
from clan_cli.inventory import Machine as InventoryMachine
from clan_cli.inventory import (
MachineDeploy,
dataclass_to_dict,
load_inventory_json,
merge_template_inventory,
set_inventory,
@@ -64,15 +63,17 @@ def create_machine(opts: CreateOptions) -> None:
clan_dir = opts.clan_dir.path
log.debug(f"Importing machine '{opts.template_name}' from {opts.template_src}")
if opts.template_name in list_nixos_machines(clan_dir) and not opts.machine.name:
machine_name = opts.machine.get("name")
if opts.template_name in list_nixos_machines(clan_dir) and not opts.machine.get(
"name"
):
msg = f"{opts.template_name} is already defined in {clan_dir}"
description = (
"Please add the --rename option to import the machine with a different name"
)
raise ClanError(msg, description=description)
machine_name = opts.template_name if not opts.machine.name else opts.machine.name
machine_name = machine_name if machine_name else opts.template_name
dst = clan_dir / "machines" / machine_name
# TODO: Move this into nix code
@@ -138,19 +139,24 @@ def create_machine(opts: CreateOptions) -> None:
merge_template_inventory(inventory, template_inventory, machine_name)
deploy = MachineDeploy()
deploy.targetHost = opts.target_host
target_host = opts.target_host
if target_host:
deploy["targetHost"] = target_host
# TODO: We should allow the template to specify machine metadata if not defined by user
new_machine = InventoryMachine(
name=machine_name, deploy=deploy, tags=opts.machine.tags
name=machine_name, deploy=deploy, tags=opts.machine.get("tags", [])
)
if (
not has_inventory
and len(opts.machine.tags) == 0
and new_machine.deploy.targetHost is None
and len(opts.machine.get("tags", [])) == 0
and new_machine.get("deploy", {}).get("targetHost") is None
):
# no need to update inventory if there are no tags or target host
return
inventory.machines.update({new_machine.name: dataclass_to_dict(new_machine)})
inventory["machines"] = inventory.get("machines", {})
inventory["machines"][machine_name] = new_machine
set_inventory(inventory, clan_dir, "Imported machine from template")

View File

@@ -13,7 +13,7 @@ from clan_cli.inventory import load_inventory_json, set_inventory
def delete_machine(flake: FlakeId, name: str) -> None:
inventory = load_inventory_json(flake.path)
machine = inventory.machines.pop(name, None)
machine = inventory.get("machines", {}).pop(name, None)
if machine is None:
msg = f"Machine {name} does not exist"
raise ClanError(msg)

View File

@@ -34,7 +34,7 @@ def set_machine(flake_url: Path, machine_name: str, machine: Machine) -> None:
@API.register
def list_inventory_machines(flake_url: str | Path) -> dict[str, Machine]:
inventory = load_inventory_eval(flake_url)
return inventory.machines
return inventory.get("machines", {})
@dataclass
@@ -61,7 +61,7 @@ def extract_header(c: str) -> str:
@API.register
def get_inventory_machine_details(flake_url: Path, machine_name: str) -> MachineDetails:
inventory = load_inventory_eval(flake_url)
machine = inventory.machines.get(machine_name)
machine = inventory.get("machines", {}).get(machine_name)
if machine is None:
msg = f"Machine {machine_name} not found in inventory"
raise ClanError(msg)
@@ -113,12 +113,12 @@ class ConnectionOptions:
def check_machine_online(
flake_url: str | Path, machine_name: str, opts: ConnectionOptions | None
) -> Literal["Online", "Offline"]:
machine = load_inventory_eval(flake_url).machines.get(machine_name)
machine = load_inventory_eval(flake_url).get("machines", {}).get(machine_name)
if not machine:
msg = f"Machine {machine_name} not found in inventory"
raise ClanError(msg)
hostname = machine.deploy.targetHost
hostname = machine.get("deploy", {}).get("targetHost")
if not hostname:
msg = f"Machine {machine_name} does not specify a targetHost"

View File

@@ -96,15 +96,19 @@ def update_machines(base_path: str, machines: list[InventoryMachine]) -> None:
# Convert InventoryMachine to Machine
for machine in machines:
name = machine.get("name")
if not name:
msg = "Machine name is not set"
raise ClanError(msg)
m = Machine(
name=machine.name,
name,
flake=FlakeId(base_path),
)
if not machine.deploy.targetHost:
msg = f"'TargetHost' is not set for machine '{machine.name}'"
if not machine.get("deploy", {}).get("targetHost"):
msg = f"'TargetHost' is not set for machine '{name}'"
raise ClanError(msg)
# Copy targetHost to machine
m.override_target_host = machine.deploy.targetHost
m.override_target_host = machine.get("deploy", {}).get("targetHost")
# Would be nice to have?
# m.override_build_host = machine.deploy.buildHost
group_machines.append(m)