Merge pull request 'Clan-app/api: add traceback for all underlying exceptions' (#2517) from hsjobeki/clan-core:hsjobeki-main into main

pkgs/clan-app/clan_app/views/webview.py

@@ -1,5 +1,6 @@
 import json
 import logging
+import traceback
 from pathlib import Path
 from typing import Any
 
@@ -107,6 +108,7 @@ class WebExecutor(GObject.Object):
             )
             return
 
+        try:
             # Initialize dataclasses from the payload
             reconciled_arguments = {}
             for k, v in data.items():
@@ -121,6 +123,22 @@ class WebExecutor(GObject.Object):
                 reconciled_arguments[k] = from_dict(arg_class, v)
 
             GLib.idle_add(fn_instance.internal_async_run, reconciled_arguments)
+        except Exception as e:
+            self.return_data_to_js(
+                method_name,
+                json.dumps(
+                    {
+                        "op_key": data["op_key"],
+                        "status": "error",
+                        "errors": [
+                            {
+                                "message": "Internal API Error",
+                                "description": traceback.format_exception(e),
+                            }
+                        ],
+                    }
+                ),
+            )
 
     def on_result(self, source: ImplFunc, data: GResult) -> None:
         result = dataclass_to_dict(data.result)

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

@@ -228,8 +228,17 @@ API.register(open_file)
         from inspect import signature
 
         func = self._registry.get(method_name, None)
-        if func:
+        if not func:
+            msg = f"API Method {method_name} not found in registry. Available methods: {list(self._registry.keys())}"
+            raise ClanError(msg)
+
         sig = signature(func)
+
+        # seems direct 'key in dict' doesnt work here
+        if arg_name not in sig.parameters.keys():  # noqa: SIM118
+            msg = f"Argument {arg_name} not found in api method '{method_name}'. Avilable arguments: {list(sig.parameters.keys())}"
+            raise ClanError(msg)
+
         param = sig.parameters.get(arg_name)
         if param:
             param_class = param.annotation

pkgs/clan-cli/clan_cli/api/disk.py

@@ -1,34 +1,142 @@
-def get_instance_name(machine_name: str) -> str:
-    return f"{machine_name}-single-disk"
+import json
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from clan_cli.api import API
+from clan_cli.dirs import TemplateType, clan_templates
+from clan_cli.errors import ClanError
+from clan_cli.machines.hardware import HardwareConfig, show_machine_hardware_config
 
 
-# @API.register
-# def set_single_disk_uuid(
-#     base_path: str,
-#     machine_name: str,
-#     disk_uuid: str,
-# ) -> None:
-#     """
-#     Set the disk UUID of single disk machine
-#     """
-#     inventory = load_inventory_json(base_path)
+@dataclass
+class Placeholder:
+    # Input name for the user
+    label: str
+    options: list[str] | None
+    required: bool
 
-#     instance_name = get_instance_name(machine_name)
 
-#     single_disk_config: ServiceSingleDisk = ServiceSingleDisk(
-#         meta=ServiceMeta(name=instance_name),
-#         roles=ServiceSingleDiskRole(
-#             default=ServiceSingleDiskRoleDefault(
-#                 config=SingleDiskConfig(device=f"/dev/disk/by-id/{disk_uuid}"),
-#                 machines=[machine_name],
-#             )
-#         ),
-#     )
+@dataclass
+class DiskSchema:
+    name: str
+    placeholders: dict[str, Placeholder]
 
-#     inventory.services.single_disk[instance_name] = single_disk_config
 
-#     save_inventory(
-#         inventory,
-#         base_path,
-#         f"Set disk UUID: '{disk_uuid}' on machine: '{machine_name}'",
-#     )
+log = logging.getLogger(__name__)
+
+
+def disk_in_facter_report(hw_report: dict) -> bool:
+    return "hardware" in hw_report and "disk" in hw_report["hardware"]
+
+
+def get_best_unix_device_name(unix_device_names: list[str]) -> str:
+    # find the first device name that is disk/by-id
+    for device_name in unix_device_names:
+        if "disk/by-id" in device_name:
+            return device_name
+    else:
+        # if no by-id found, use the first device name
+        return unix_device_names[0]
+
+
+def hw_main_disk_options(hw_report: dict) -> list[str]:
+    if not disk_in_facter_report(hw_report):
+        msg = "hw_report doesnt include 'disk' information"
+        raise ClanError(msg, description=f"{hw_report.keys()}")
+
+    disks = hw_report["hardware"]["disk"]
+
+    options: list[str] = []
+    for disk in disks:
+        unix_device_names = disk["unix_device_names"]
+        device_name = get_best_unix_device_name(unix_device_names)
+        options += [device_name]
+
+    return options
+
+
+# must be manually kept in sync with the ${clancore}/templates/disks directory
+templates: dict[str, dict[str, Callable[[dict[str, Any]], Placeholder]]] = {
+    "single-disk": {
+        # Placeholders
+        "mainDisk": lambda hw_report: Placeholder(
+            label="Main disk", options=hw_main_disk_options(hw_report), required=True
+        ),
+    }
+}
+
+
+@API.register
+def get_disk_schemas(base_path: Path, machine_name: str) -> dict[str, DiskSchema]:
+    """
+    Get the available disk schemas
+    """
+    disk_templates = clan_templates(TemplateType.DISK)
+    disk_schemas = {}
+
+    hw_report_path = HardwareConfig.NIXOS_FACTER.config_path(base_path, machine_name)
+    if not hw_report_path.exists():
+        msg = "Hardware configuration missing"
+        raise ClanError(msg)
+
+    hw_report = {}
+    with hw_report_path.open("r") as hw_report_file:
+        hw_report = json.load(hw_report_file)
+
+    for disk_template in disk_templates.iterdir():
+        if disk_template.is_file():
+            schema_name = disk_template.stem
+            if schema_name not in templates:
+                msg = f"Disk schema {schema_name} not found in templates"
+                raise ClanError(
+                    msg,
+                    description="This is an internal architecture problem. Because disk schemas dont define their own interface",
+                )
+
+            placeholder_getters = templates.get(schema_name)
+            placeholders = {}
+
+            if placeholder_getters:
+                placeholders = {k: v(hw_report) for k, v in placeholder_getters.items()}
+
+            disk_schemas[schema_name] = DiskSchema(
+                name=schema_name, placeholders=placeholders
+            )
+
+    return disk_schemas
+
+
+@API.register
+def set_machine_disk_schema(
+    base_path: Path,
+    machine_name: str,
+    schema_name: str,
+    # Placeholders are used to fill in the disk schema
+    placeholders: dict[str, str],
+) -> None:
+    """
+    Set the disk placeholders of the template
+    """
+    # Assert the hw-config must exist before setting the disk
+    hw_config = show_machine_hardware_config(base_path, machine_name)
+    hw_config_path = hw_config.config_path(base_path, machine_name)
+
+    if not hw_config_path.exists():
+        msg = "Hardware configuration must exist before applying disk schema"
+        raise ClanError(msg)
+
+    if hw_config != HardwareConfig.NIXOS_FACTER:
+        msg = "Hardware configuration must use type FACTER for applying disk schema automatically"
+        raise ClanError(msg)
+
+    disk_schema_path = clan_templates(TemplateType.DISK) / f"{schema_name}.nix"
+
+    if not disk_schema_path.exists():
+        msg = f"Disk schema not found at {disk_schema_path}"
+        raise ClanError(msg)
+
+    with disk_schema_path.open("r") as disk_template:
+        print(disk_template.read())

pkgs/clan-cli/clan_cli/api/serde.py

@@ -160,7 +160,7 @@ def construct_value(
     if loc is None:
         loc = []
     if t is None and field_value:
-        msg = f"Expected None but got: {field_value}"
+        msg = f"Trying to construct field of type None. But got: {field_value}. loc: {loc}"
         raise ClanError(msg, location=f"{loc}")
 
     if is_type_in_union(t, type(None)) and field_value is None:
@@ -319,4 +319,5 @@ def from_dict(
             msg = f"{data} is not a dict. Expected {t}"
             raise ClanError(msg)
         return construct_dataclass(t, data, path)  # type: ignore
+    # breakpoint()
     return construct_value(t, data, path)

pkgs/clan-cli/clan_cli/dirs.py

@@ -48,6 +48,7 @@ def find_toplevel(top_level_files: list[str]) -> Path | None:
 
 class TemplateType(Enum):
     CLAN = "clan"
+    DISK = "disk"
 
 
 def clan_templates(template_type: TemplateType) -> Path:

templates/disk/single-disk.nix

@@ -25,12 +25,6 @@
                 mountpoint = "/";
               };
             };
-            swap = {
-              size = "{{swapSize}}";
-              content = {
-                type = "swap";
-              };
-            };
           };
         };
       };