inventory.{cli,api}: use only dictionaries

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

View File

@@ -272,11 +272,14 @@ def set_service_instance(
inventory = load_inventory_json(base_path)
target_type = get_args(get_type_hints(Service)[module_name])[1]
module_instance_map: dict[str, Any] = getattr(inventory.services, module_name, {})
module_instance_map: dict[str, Any] = inventory.get("services", {}).get(
module_name, {}
)
module_instance_map[instance_name] = from_dict(target_type, config)
setattr(inventory.services, module_name, module_instance_map)
inventory["services"] = inventory.get("services", {})
inventory["services"][module_name] = module_instance_map
set_inventory(
inventory, base_path, f"Update {module_name} instance {instance_name}"

View File

@@ -63,9 +63,11 @@ def show_clan_meta(uri: str | Path) -> Meta:
)
return Meta(
name=clan_meta.get("name"),
description=clan_meta.get("description", None),
icon=icon_path,
{
"name": clan_meta.get("name"),
"description": clan_meta.get("description"),
"icon": icon_path if icon_path else "",
}
)
@@ -73,9 +75,9 @@ def show_command(args: argparse.Namespace) -> None:
flake_path = args.flake.path
meta = show_clan_meta(flake_path)
print(f"Name: {meta.name}")
print(f"Description: {meta.description or '-'}")
print(f"Icon: {meta.icon or '-'}")
print(f"Name: {meta.get("name")}")
print(f"Description: {meta.get("description", '-')}")
print(f"Icon: {meta.get("icon", '-')}")
def register_parser(parser: argparse.ArgumentParser) -> None:

View File

@@ -1,7 +1,7 @@
from dataclasses import dataclass
from clan_cli.api import API
from clan_cli.inventory import Meta, load_inventory_json, set_inventory
from clan_cli.inventory import Inventory, Meta, load_inventory_json, set_inventory
@dataclass
@@ -11,10 +11,10 @@ class UpdateOptions:
@API.register
def update_clan_meta(options: UpdateOptions) -> Meta:
def update_clan_meta(options: UpdateOptions) -> Inventory:
inventory = load_inventory_json(options.directory)
inventory.meta = options.meta
inventory["meta"] = options.meta
set_inventory(inventory, options.directory, "Update clan metadata")
return inventory.meta
return inventory

View File

@@ -61,9 +61,7 @@ def get_inventory_path(flake_dir: str | Path, create: bool = True) -> Path:
# Default inventory
default_inventory = Inventory(
meta=Meta(name="New Clan"), machines={}, services=Service()
)
default_inventory: Inventory = {"meta": {"name": "New Clan"}}
@API.register
@@ -381,9 +379,7 @@ def patch_inventory_with(base_dir: Path, section: str, content: dict[str, Any])
@API.register
def set_inventory(
inventory: Inventory | dict[str, Any], flake_dir: str | Path, message: str
) -> None:
def set_inventory(inventory: Inventory, flake_dir: str | Path, message: str) -> None:
"""
Write the inventory to the flake directory
and commit it to git with the given message
@@ -396,18 +392,11 @@ def set_inventory(
filtered_modules = lambda m: {
key: value for key, value in m.items() if "/nix/store" not in value
}
if isinstance(inventory, dict):
modules = filtered_modules(inventory.get("modules", {})) # type: ignore
inventory["modules"] = modules
else:
modules = filtered_modules(inventory.modules) # type: ignore
inventory.modules = modules
modules = filtered_modules(inventory.get("modules", {})) # type: ignore
inventory["modules"] = modules
with inventory_file.open("w") as f:
if isinstance(inventory, Inventory):
json.dump(dataclass_to_dict(inventory), f, indent=2)
else:
json.dump(inventory, f, indent=2)
json.dump(inventory, f, indent=2)
commit_file(inventory_file, Path(flake_dir), commit_message=message)
@@ -436,7 +425,7 @@ def merge_template_inventory(
Merge the template inventory into the current inventory
The template inventory is expected to be a subset of the current inventory
"""
for service_name, instance in template_inventory.services.items():
for service_name, instance in template_inventory.get("services", {}).items():
if len(instance.keys()) > 0:
msg = f"Service {service_name} in template inventory has multiple instances"
description = (

View File

@@ -5,7 +5,7 @@
# ruff: noqa: N806
# ruff: noqa: F401
# fmt: off
from typing import Any, Literal, TypedDict, NotRequired
from typing import Any, Literal, NotRequired, TypedDict
class MachineDeploy(TypedDict):

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)

View File

@@ -0,0 +1,41 @@
from clan_cli.inventory.classes import Inventory, Machine, Meta, Service
def test_make_meta_minimal() -> None:
# Name is required
res = Meta(
{
"name": "foo",
}
)
assert res == {"name": "foo"}
def test_make_inventory_minimal() -> None:
# Meta is required
res = Inventory(
{
"meta": Meta(
{
"name": "foo",
}
),
}
)
assert res == {"meta": {"name": "foo"}}
def test_make_machine_minimal() -> None:
# Empty is valid
res = Machine({})
assert res == {}
def test_make_service_minimal() -> None:
# Empty is valid
res = Service({})
assert res == {}

View File

@@ -72,7 +72,7 @@ def test_add_module_to_inventory(
inventory = load_inventory_json(base_path)
inventory.services = {
inventory["services"] = {
"borgbackup": {
"borg1": {
"meta": {"name": "borg1"},