Merge pull request 'API: migrate add machine to inventory' (#1676) from hsjobeki/clan-core:hsjobeki-main into main

pkgs/clan-cli/clan_cli/inventory/__init__.py

@@ -1,4 +1,4 @@
-import re
+import json
 from dataclasses import asdict, dataclass, is_dataclass
 from pathlib import Path
 from typing import Any, Literal
@@ -55,16 +55,6 @@ class Machine:
 
     @staticmethod
     def from_dict(d: dict[str, Any]) -> "Machine":
-        if "name" not in d:
-            raise ClanError("name not found in machine")
-
-        hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
-        if not re.match(hostname_regex, d["name"]):
-            raise ClanError(
-                "Machine name must be a valid hostname",
-                description=f"""Machine name: {d["name"]}""",
-            )
-
         return Machine(**d)
 
 
@@ -94,15 +84,9 @@ class Service:
 
     @staticmethod
     def from_dict(d: dict[str, Any]) -> "Service":
-        if "meta" not in d:
-            raise ClanError("meta not found in service")
-
-        if "roles" not in d:
-            raise ClanError("roles not found in service")
-
         return Service(
-            meta=ServiceMeta(**d["meta"]),
-            roles={name: Role(**role) for name, role in d["roles"].items()},
+            meta=ServiceMeta(**d.get("meta", {})),
+            roles={name: Role(**role) for name, role in d.get("roles", {}).items()},
             machines={
                 name: MachineServiceConfig(**machine)
                 for name, machine in d.get("machines", {}).items()
@@ -117,22 +101,39 @@ class Inventory:
 
     @staticmethod
     def from_dict(d: dict[str, Any]) -> "Inventory":
-        if "machines" not in d:
-            raise ClanError("machines not found in inventory")
-
-        if "services" not in d:
-            raise ClanError("services not found in inventory")
-
         return Inventory(
             machines={
                 name: Machine.from_dict(machine)
-                for name, machine in d["machines"].items()
+                for name, machine in d.get("machines", {}).items()
             },
             services={
                 name: {
                     role: Service.from_dict(service)
                     for role, service in services.items()
                 }
-                for name, services in d["services"].items()
+                for name, services in d.get("services", {}).items()
             },
         )
+
+    @staticmethod
+    def get_path(flake_dir: str | Path) -> Path:
+        return Path(flake_dir) / "inventory.json"
+
+    @staticmethod
+    def load_file(flake_dir: str | Path) -> "Inventory":
+        inventory = Inventory(machines={}, services={})
+        inventory_file = Inventory.get_path(flake_dir)
+        if inventory_file.exists():
+            with open(inventory_file) as f:
+                try:
+                    res = json.load(f)
+                    inventory = Inventory.from_dict(res)
+                except json.JSONDecodeError as e:
+                    raise ClanError(f"Error decoding inventory file: {e}")
+
+        return inventory
+
+    def persist(self, flake_dir: str | Path) -> None:
+        inventory_file = Inventory.get_path(flake_dir)
+        with open(inventory_file, "w") as f:
+            json.dump(dataclass_to_dict(self), f, indent=2)

pkgs/clan-cli/clan_cli/machines/create.py

@@ -1,30 +1,71 @@
 import argparse
 import logging
-from dataclasses import dataclass
+import re
 from pathlib import Path
-from typing import Any
 
 from clan_cli.api import API
-from clan_cli.config.machine import set_config_for_machine
+from clan_cli.errors import ClanError
+from clan_cli.git import commit_file
+from clan_cli.inventory import Inventory, Machine
 
 log = logging.getLogger(__name__)
 
 
-@dataclass
-class MachineCreateRequest:
-    name: str
-    config: dict[str, Any]
-
-
 @API.register
-def create_machine(flake_dir: str | Path, machine: MachineCreateRequest) -> None:
-    set_config_for_machine(Path(flake_dir), machine.name, machine.config)
+def create_machine(flake_dir: str | Path, machine: Machine) -> None:
+    hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
+    if not re.match(hostname_regex, machine.name):
+        raise ClanError(
+            "Machine name must be a valid hostname", location="Create Machine"
+        )
+
+    inventory = Inventory.load_file(flake_dir)
+    inventory.machines.update({machine.name: machine})
+    inventory.persist(flake_dir)
+
+    commit_file(Inventory.get_path(flake_dir), Path(flake_dir))
 
 
 def create_command(args: argparse.Namespace) -> None:
-    create_machine(args.flake, MachineCreateRequest(args.machine, dict()))
+    create_machine(
+        args.flake,
+        Machine(
+            name=args.machine,
+            system=args.system,
+            description=args.description,
+            tags=args.tags,
+            icon=args.icon,
+        ),
+    )
 
 
 def register_create_parser(parser: argparse.ArgumentParser) -> None:
     parser.add_argument("machine", type=str)
     parser.set_defaults(func=create_command)
+
+    parser.add_argument(
+        "--system",
+        type=str,
+        default=None,
+        help="Host platform to use. i.e. 'x86_64-linux' or 'aarch64-darwin' etc.",
+        metavar="PLATFORM",
+    )
+    parser.add_argument(
+        "--description",
+        type=str,
+        default=None,
+        help="A description of the machine.",
+    )
+    parser.add_argument(
+        "--icon",
+        type=str,
+        default=None,
+        help="Path to an icon to use for the machine. - Must be a path to icon file relative to the flake directory, or a public url.",
+        metavar="PATH",
+    )
+    parser.add_argument(
+        "--tags",
+        nargs="+",
+        default=[],
+        help="Tags to associate with the machine. Can be used to assign multiple machines to services.",
+    )

pkgs/clan-cli/clan_cli/machines/delete.py

@@ -1,21 +1,36 @@
 import argparse
 import shutil
+from pathlib import Path
+
+from clan_cli.api import API
+from clan_cli.inventory import Inventory
 
 from ..completions import add_dynamic_completer, complete_machines
 from ..dirs import specific_machine_dir
 from ..errors import ClanError
 
 
-def delete_command(args: argparse.Namespace) -> None:
-    folder = specific_machine_dir(args.flake, args.host)
+@API.register
+def delete_machine(base_dir: str | Path, name: str) -> None:
+    inventory = Inventory.load_file(base_dir)
+
+    machine = inventory.machines.pop(name, None)
+    if machine is None:
+        raise ClanError(f"Machine {name} does not exist")
+
+    inventory.persist(base_dir)
+
+    folder = specific_machine_dir(Path(base_dir), name)
     if folder.exists():
         shutil.rmtree(folder)
-    else:
-        raise ClanError(f"Machine {args.host} does not exist")
+
+
+def delete_command(args: argparse.Namespace) -> None:
+    delete_machine(args.flake, args.name)
 
 
 def register_delete_parser(parser: argparse.ArgumentParser) -> None:
-    machines_parser = parser.add_argument("host", type=str)
+    machines_parser = parser.add_argument("name", type=str)
     add_dynamic_completer(machines_parser, complete_machines)
 
     parser.set_defaults(func=delete_command)

pkgs/clan-cli/clan_cli/machines/list.py

@@ -4,22 +4,19 @@ import logging
 from pathlib import Path
 
 from clan_cli.api import API
+from clan_cli.inventory import Machine
 
 from ..cmd import run_no_stdout
-from ..nix import nix_config, nix_eval
+from ..nix import nix_eval
 
 log = logging.getLogger(__name__)
 
 
 @API.register
-def list_machines(flake_url: str | Path, debug: bool = False) -> list[str]:
-    config = nix_config()
-    system = config["system"]
+def list_machines(flake_url: str | Path, debug: bool = False) -> dict[str, Machine]:
     cmd = nix_eval(
         [
-            f"{flake_url}#clanInternals.machines.{system}",
-            "--apply",
-            "builtins.attrNames",
+            f"{flake_url}#clanInternals.inventory.machines",
             "--json",
         ]
     )
@@ -27,12 +24,13 @@ def list_machines(flake_url: str | Path, debug: bool = False) -> list[str]:
     proc = run_no_stdout(cmd)
 
     res = proc.stdout.strip()
-    return json.loads(res)
+    data = {name: Machine.from_dict(v) for name, v in json.loads(res).items()}
+    return data
 
 
 def list_command(args: argparse.Namespace) -> None:
     flake_path = Path(args.flake).resolve()
-    for name in list_machines(flake_path, args.debug):
+    for name in list_machines(flake_path, args.debug).keys():
         print(name)
 
 

pkgs/clan-cli/tests/test_machines_config.py

@@ -7,7 +7,8 @@ from clan_cli.config.machine import (
     verify_machine_config,
 )
 from clan_cli.config.schema import machine_schema
-from clan_cli.machines.create import MachineCreateRequest, create_machine
+from clan_cli.inventory import Machine
+from clan_cli.machines.create import create_machine
 from clan_cli.machines.list import list_machines
 
 
@@ -19,17 +20,24 @@ def test_schema_for_machine(test_flake_with_core: FlakeForTest) -> None:
 
 @pytest.mark.with_core
 def test_create_machine_on_minimal_clan(test_flake_minimal: FlakeForTest) -> None:
-    assert list_machines(test_flake_minimal.path) == []
+    assert list_machines(test_flake_minimal.path) == {}
     create_machine(
         test_flake_minimal.path,
-        MachineCreateRequest(
-            name="foo", config=dict(nixpkgs=dict(hostSystem="x86_64-linux"))
+        Machine(
+            name="foo",
+            system="x86_64-linux",
+            description="A test machine",
+            tags=["test"],
+            icon=None,
         ),
     )
-    assert list_machines(test_flake_minimal.path) == ["foo"]
+    assert list(list_machines(test_flake_minimal.path).keys()) == ["foo"]
+
+    # Writes into settings.json
     set_config_for_machine(
         test_flake_minimal.path, "foo", dict(services=dict(openssh=dict(enable=True)))
     )
+
     config = config_for_machine(test_flake_minimal.path, "foo")
     assert config["services"]["openssh"]["enable"]
     assert verify_machine_config(test_flake_minimal.path, "foo") is None