Merge pull request 'Init: Autogenerate classes from nix interfaces' (#1778) from hsjobeki/clan-core:hsjobeki-main into main

.gitignore

@@ -14,6 +14,7 @@ nixos.qcow2
 **/*.glade~
 /docs/out
 
+
 # dream2nix
 .dream2nix
 

clanModules/single-disk/default.nix

@@ -2,7 +2,8 @@
 {
   options.clan.single-disk = {
     device = lib.mkOption {
-      type = lib.types.str;
+      default = null;
+      type = lib.types.nullOr lib.types.str;
       description = "The primary disk device to install the system on";
       # Question: should we set a default here?
       # default = "/dev/null";

inventory.json

@@ -26,9 +26,6 @@
         },
         "roles": {
           "default": {
-            "config": {
-              "packages": ["vim"]
-            },
             "imports": [],
             "machines": ["test-inventory-machine"],
             "tags": []

lib/inventory/build-inventory/interface.nix

@@ -41,6 +41,7 @@ in
           options = {
             inherit (metaOptions) name description icon;
             tags = lib.mkOption {
+
               default = [ ];
               apply = lib.unique;
               type = types.listOf types.str;
@@ -49,16 +50,10 @@ in
               default = null;
               type = types.nullOr types.str;
             };
-            deploy = lib.mkOption {
-              default = { };
-              type = types.submodule {
-                options = {
-                  targetHost = lib.mkOption {
-                    default = null;
-                    type = types.nullOr types.str;
-                  };
-                };
-              };
+            deploy.targetHost = lib.mkOption {
+              description = "Configuration for the deployment of the machine";
+              default = null;
+              type = types.nullOr types.str;
             };
           };
         }

lib/inventory/interface-to-schema.nix

@@ -55,6 +55,7 @@ let
               inventorySchema.properties.services.additionalProperties.additionalProperties.properties.meta;
             config = {
               title = "${moduleName}-config";
+              default = { };
             } // moduleSchema;
             roles = {
               type = "object";
@@ -69,6 +70,7 @@ let
                       {
                         properties.config = {
                           title = "${moduleName}-config";
+                          default = { };
                         } // moduleSchema;
                       };
                 }) (rolesOf moduleName)
@@ -80,6 +82,7 @@ let
                 {
                   additionalProperties.properties.config = {
                     title = "${moduleName}-config";
+                    default = { };
                   } // moduleSchema;
                 };
           };

lib/jsonschema/default.nix

@@ -318,7 +318,7 @@ rec {
     # return jsonschema property definition for submodule
     # then (lib.attrNames (option.type.getSubOptions option.loc).opt)
     then
-      parseOptions' (option.type.getSubOptions option.loc)
+      example // description // parseOptions' (option.type.getSubOptions option.loc)
     # throw error if option type is not supported
     else
       notSupported option;

lib/jsonschema/test_parseOption.nix

@@ -279,6 +279,7 @@ in
       expected = {
         type = "object";
         additionalProperties = false;
+        description = "Test Description";
         properties = {
           opt = {
             type = "boolean";
@@ -303,6 +304,7 @@ in
       expected = {
         type = "object";
         additionalProperties = false;
+        description = "Test Description";
         properties = {
           opt = {
             type = "boolean";

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

@@ -6,7 +6,7 @@ from pathlib import Path
 
 from clan_cli.cmd import run_no_stdout
 from clan_cli.errors import ClanCmdError, ClanError
-from clan_cli.inventory import Inventory, Service
+from clan_cli.inventory import Inventory, load_inventory
 from clan_cli.nix import nix_eval
 
 from . import API
@@ -150,24 +150,6 @@ def get_module_info(
     )
 
 
-@API.register
-def update_module_instance(
-    base_path: str, module_name: str, instance_name: str, instance_config: Service
-) -> Inventory:
-    inventory = Inventory.load_file(base_path)
-
-    module_instances = inventory.services.get(module_name, {})
-    module_instances[instance_name] = instance_config
-
-    inventory.services[module_name] = module_instances
-
-    inventory.persist(
-        base_path, f"Updated module instance {module_name}/{instance_name}"
-    )
-
-    return inventory
-
-
 @API.register
 def get_inventory(base_path: str) -> Inventory:
-    return Inventory.load_file(base_path)
+    return load_inventory(base_path)

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

@@ -6,7 +6,7 @@ from pathlib import Path
 
 from clan_cli.api import API
 from clan_cli.arg_actions import AppendOptionAction
-from clan_cli.inventory import Inventory, InventoryMeta
+from clan_cli.inventory import Meta, load_inventory, save_inventory
 
 from ..cmd import CmdOut, run
 from ..errors import ClanError
@@ -29,7 +29,7 @@ class CreateOptions:
     directory: Path | str
     # Metadata for the clan
     # Metadata can be shown with `clan show`
-    meta: InventoryMeta | None = None
+    meta: Meta | None = None
     # URL to the template to use. Defaults to the "minimal" template
     template_url: str = minimal_template_url
 
@@ -84,11 +84,11 @@ def create_clan(options: CreateOptions) -> CreateClanResponse:
     ## End: setup git
 
     # Write inventory.json file
-    inventory = Inventory.load_file(directory)
+    inventory = load_inventory(directory)
     if options.meta is not None:
         inventory.meta = options.meta
     # Persist creates a commit message for each change
-    inventory.persist(directory, "Init inventory")
+    save_inventory(inventory, directory, "Init inventory")
 
     command = ["nix", "flake", "update"]
     out = run(command, cwd=directory)
@@ -113,7 +113,7 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
 
     parser.add_argument(
         "--meta",
-        help=f"""Metadata to set for the clan. Available options are: {", ".join([f.name for f in fields(InventoryMeta)]) }""",
+        help=f"""Metadata to set for the clan. Available options are: {", ".join([f.name for f in fields(Meta)]) }""",
         nargs=2,
         metavar=("name", "value"),
         action=AppendOptionAction,

pkgs/clan-cli/clan_cli/clan/show.py

@@ -6,7 +6,7 @@ from urllib.parse import urlparse
 
 from clan_cli.api import API
 from clan_cli.errors import ClanCmdError, ClanError
-from clan_cli.inventory import InventoryMeta
+from clan_cli.inventory import Meta
 
 from ..cmd import run_no_stdout
 from ..nix import nix_eval
@@ -15,7 +15,7 @@ log = logging.getLogger(__name__)
 
 
 @API.register
-def show_clan_meta(uri: str | Path) -> InventoryMeta:
+def show_clan_meta(uri: str | Path) -> Meta:
     cmd = nix_eval(
         [
             f"{uri}#clanInternals.inventory.meta",
@@ -61,7 +61,7 @@ def show_clan_meta(uri: str | Path) -> InventoryMeta:
                 description="Icon path must be a URL or a relative path.",
             )
 
-    return InventoryMeta(
+    return Meta(
         name=clan_meta.get("name"),
         description=clan_meta.get("description", None),
         icon=icon_path,

pkgs/clan-cli/clan_cli/clan/update.py

@@ -1,20 +1,20 @@
 from dataclasses import dataclass
 
 from clan_cli.api import API
-from clan_cli.inventory import Inventory, InventoryMeta
+from clan_cli.inventory import Meta, load_inventory, save_inventory
 
 
 @dataclass
 class UpdateOptions:
     directory: str
-    meta: InventoryMeta
+    meta: Meta
 
 
 @API.register
-def update_clan_meta(options: UpdateOptions) -> InventoryMeta:
-    inventory = Inventory.load_file(options.directory)
+def update_clan_meta(options: UpdateOptions) -> Meta:
+    inventory = load_inventory(options.directory)
     inventory.meta = options.meta
 
-    inventory.persist(options.directory, "Update clan meta")
+    save_inventory(inventory, options.directory, "Update clan metadata")
 
     return inventory.meta

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

@@ -1,16 +1,44 @@
-# ruff: noqa: N815
-# ruff: noqa: N806
+import dataclasses
 import json
-from dataclasses import asdict, dataclass, field, is_dataclass
+from dataclasses import fields, is_dataclass
 from pathlib import Path
-from typing import Any, Literal
+from types import UnionType
+from typing import Any, get_args, get_origin
 
 from clan_cli.errors import ClanError
 from clan_cli.git import commit_file
 
+from .classes import (
+    Inventory,
+    Machine,
+    MachineDeploy,
+    Meta,
+    Service,
+    ServiceBorgbackup,
+    ServiceBorgbackupMeta,
+    ServiceBorgbackupRole,
+    ServiceBorgbackupRoleClient,
+    ServiceBorgbackupRoleServer,
+)
+
+# Re export classes here
+# This allows to rename classes in the generated code
+__all__ = [
+    "Service",
+    "Machine",
+    "Meta",
+    "Inventory",
+    "MachineDeploy",
+    "ServiceBorgbackup",
+    "ServiceBorgbackupMeta",
+    "ServiceBorgbackupRole",
+    "ServiceBorgbackupRoleClient",
+    "ServiceBorgbackupRoleServer",
+]
+
 
 def sanitize_string(s: str) -> str:
-    return s.replace("\\", "\\\\").replace('"', '\\"')
+    return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
 
 
 def dataclass_to_dict(obj: Any) -> Any:
@@ -22,8 +50,11 @@ def dataclass_to_dict(obj: Any) -> Any:
     """
     if is_dataclass(obj):
         return {
-            sanitize_string(k): dataclass_to_dict(v)
-            for k, v in asdict(obj).items()  # type: ignore
+            # Use either the original name or name
+            sanitize_string(
+                field.metadata.get("original_name", field.name)
+            ): dataclass_to_dict(getattr(obj, field.name))
+            for field in fields(obj)  # type: ignore
         }
     elif isinstance(obj, list | tuple):
         return [dataclass_to_dict(item) for item in obj]
@@ -37,149 +68,133 @@ def dataclass_to_dict(obj: Any) -> Any:
         return obj
 
 
-@dataclass
-class DeploymentInfo:
+def is_union_type(type_hint: type) -> bool:
+    return type(type_hint) is UnionType
+
+
+def get_inner_type(type_hint: type) -> type:
+    if is_union_type(type_hint):
+        # Return the first non-None type
+        return next(t for t in get_args(type_hint) if t is not type(None))
+    return type_hint
+
+
+def get_second_type(type_hint: type[dict]) -> type:
     """
-    Deployment information for a machine.
+    Get the value type of a dictionary type hint
     """
+    args = get_args(type_hint)
+    if len(args) == 2:
+        # Return the second argument, which should be the value type (Machine)
+        return args[1]
 
-    targetHost: str | None = None
+    raise ValueError(f"Invalid type hint for dict: {type_hint}")
 
 
-@dataclass
-class Machine:
+def from_dict(t: type, data: dict[str, Any] | None) -> Any:
     """
-    Inventory machine model.
-
-    DO NOT EDIT THIS CLASS.
-    Any changes here must be reflected in the inventory interface file and potentially other nix files.
-
-    - Persisted to the inventory.json file
-    - Source of truth to generate each clan machine.
-    - For hardware deployment, the machine must declare the host system.
+    Dynamically instantiate a data class from a dictionary, handling nested data classes.
     """
+    if data is None:
+        return None
 
-    name: str
-    deploy: DeploymentInfo = field(default_factory=DeploymentInfo)
-    description: str | None = None
-    icon: str | None = None
-    tags: list[str] = field(default_factory=list)
-    system: Literal["x86_64-linux"] | str | None = None
+    try:
+        # Attempt to create an instance of the data_class
+        field_values = {}
+        for field in fields(t):
+            original_name = field.metadata.get("original_name", field.name)
 
-    @staticmethod
-    def from_dict(data: dict[str, Any]) -> "Machine":
-        targetHost = data.get("deploy", {}).get("targetHost", None)
-        return Machine(
-            name=data["name"],
-            description=data.get("description", None),
-            icon=data.get("icon", None),
-            tags=data.get("tags", []),
-            system=data.get("system", None),
-            deploy=DeploymentInfo(targetHost),
-        )
+            field_value = data.get(original_name)
+
+            field_type = get_inner_type(field.type)  # type: ignore
+
+            if original_name in data:
+                # If the field is another dataclass, recursively instantiate it
+                if is_dataclass(field_type):
+                    field_value = from_dict(field_type, field_value)
+                elif isinstance(field_type, Path | str) and isinstance(
+                    field_value, str
+                ):
+                    field_value = (
+                        Path(field_value) if field_type == Path else field_value
+                    )
+                elif get_origin(field_type) is dict and isinstance(field_value, dict):
+                    # The field is a dictionary with a specific type
+                    inner_type = get_second_type(field_type)
+                    field_value = {
+                        k: from_dict(inner_type, v) for k, v in field_value.items()
+                    }
+                elif get_origin is list and isinstance(field_value, list):
+                    # The field is a list with a specific type
+                    inner_type = get_args(field_type)[0]
+                    field_value = [from_dict(inner_type, v) for v in field_value]
+
+            # Set the value
+            if (
+                field.default is not dataclasses.MISSING
+                or field.default_factory is not dataclasses.MISSING
+            ):
+                # Fields with default value
+                # a: Int = 1
+                # b: list = Field(default_factory=list)
+                if original_name in data or field_value is not None:
+                    field_values[field.name] = field_value
+            else:
+                # Fields without default value
+                # a: Int
+                field_values[field.name] = field_value
+
+        return t(**field_values)
+
+    except (TypeError, ValueError) as e:
+        print(f"Failed to instantiate {t.__name__}: {e} {data}")
+        return None
+        # raise ClanError(f"Failed to instantiate {t.__name__}: {e}")
 
 
-@dataclass
-class MachineServiceConfig:
-    config: dict[str, Any] = field(default_factory=dict)
-    imports: list[str] = field(default_factory=list)
+def get_path(flake_dir: str | Path) -> Path:
+    """
+    Get the path to the inventory file in the flake directory
+    """
+    return (Path(flake_dir) / "inventory.json").resolve()
 
 
-@dataclass
-class ServiceMeta:
-    name: str
-    description: str | None = None
-    icon: str | None = None
+# Default inventory
+default_inventory = Inventory(
+    meta=Meta(name="New Clan"), machines={}, services=Service()
+)
 
 
-@dataclass
-class Role:
-    config: dict[str, Any] = field(default_factory=dict)
-    imports: list[str] = field(default_factory=list)
-    machines: list[str] = field(default_factory=list)
-    tags: list[str] = field(default_factory=list)
+def load_inventory(
+    flake_dir: str | Path, default: Inventory = default_inventory
+) -> Inventory:
+    """
+    Load the inventory file from the flake directory
+    If not file is found, returns the default inventory
+    """
+    inventory = default_inventory
+
+    inventory_file = get_path(flake_dir)
+    if inventory_file.exists():
+        with open(inventory_file) as f:
+            try:
+                res = json.load(f)
+                inventory = from_dict(Inventory, res)
+            except json.JSONDecodeError as e:
+                # Error decoding the inventory file
+                raise ClanError(f"Error decoding inventory file: {e}")
+
+    return inventory
 
 
-@dataclass
-class Service:
-    meta: ServiceMeta
-    roles: dict[str, Role]
-    config: dict[str, Any] = field(default_factory=dict)
-    imports: list[str] = field(default_factory=list)
-    machines: dict[str, MachineServiceConfig] = field(default_factory=dict)
+def save_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
+    """
+    inventory_file = get_path(flake_dir)
 
-    @staticmethod
-    def from_dict(d: dict[str, Any]) -> "Service":
-        return Service(
-            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()
-                }
-                if d.get("machines")
-                else {}
-            ),
-            config=d.get("config", {}),
-            imports=d.get("imports", []),
-        )
+    with open(inventory_file, "w") as f:
+        json.dump(dataclass_to_dict(inventory), f, indent=2)
 
-
-@dataclass
-class InventoryMeta:
-    name: str
-    description: str | None = None
-    icon: str | None = None
-
-
-@dataclass
-class Inventory:
-    meta: InventoryMeta
-    machines: dict[str, Machine]
-    services: dict[str, dict[str, Service]]
-
-    @staticmethod
-    def from_dict(d: dict[str, Any]) -> "Inventory":
-        return Inventory(
-            meta=InventoryMeta(**d.get("meta", {})),
-            machines={
-                name: Machine.from_dict(machine)
-                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.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={}, meta=InventoryMeta(name="New Clan")
-        )
-        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, message: str) -> None:
-        inventory_file = Inventory.get_path(flake_dir)
-
-        with open(inventory_file, "w") as f:
-            json.dump(dataclass_to_dict(self), f, indent=2)
-
-        commit_file(inventory_file, Path(flake_dir), commit_message=message)
+    commit_file(inventory_file, Path(flake_dir), commit_message=message)

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

@@ -0,0 +1,176 @@
+# DON NOT EDIT THIS FILE MANUALLY. IT IS GENERATED.
+#
+# ruff: noqa: N815
+# ruff: noqa: N806
+# ruff: noqa: F401
+# fmt: off
+from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class MachineDeploy:
+    targetHost: str | None = field(default=None )
+
+
+@dataclass
+class Machine:
+    deploy: MachineDeploy
+    name: str
+    description: str | None = field(default=None )
+    icon: str | None = field(default=None )
+    system: str | None = field(default=None )
+    tags: list[str] = field(default_factory=list )
+
+
+@dataclass
+class Meta:
+    name: str
+    description: str | None = field(default=None )
+    icon: str | None = field(default=None )
+
+
+@dataclass
+class BorgbackupConfigDestination:
+    repo: str
+    name: str
+
+
+@dataclass
+class BorgbackupConfig:
+    destinations: dict[str, BorgbackupConfigDestination] = field(default_factory=dict )
+
+
+@dataclass
+class ServiceBorgbackupMachine:
+    config: BorgbackupConfig = field(default_factory=BorgbackupConfig )
+    imports: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServiceBorgbackupMeta:
+    name: str
+    description: str | None = field(default=None )
+    icon: str | None = field(default=None )
+
+
+@dataclass
+class ServiceBorgbackupRoleClient:
+    config: BorgbackupConfig = field(default_factory=BorgbackupConfig )
+    imports: list[str] = field(default_factory=list )
+    machines: list[str] = field(default_factory=list )
+    tags: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServiceBorgbackupRoleServer:
+    config: BorgbackupConfig = field(default_factory=BorgbackupConfig )
+    imports: list[str] = field(default_factory=list )
+    machines: list[str] = field(default_factory=list )
+    tags: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServiceBorgbackupRole:
+    client: ServiceBorgbackupRoleClient
+    server: ServiceBorgbackupRoleServer
+
+
+@dataclass
+class ServiceBorgbackup:
+    meta: ServiceBorgbackupMeta
+    roles: ServiceBorgbackupRole
+    config: BorgbackupConfig = field(default_factory=BorgbackupConfig )
+    machines: dict[str, ServiceBorgbackupMachine] = field(default_factory=dict )
+
+
+@dataclass
+class PackagesConfig:
+    packages: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServicePackageMachine:
+    config: PackagesConfig = field(default_factory=PackagesConfig )
+    imports: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServicePackageMeta:
+    name: str
+    description: str | None = field(default=None )
+    icon: str | None = field(default=None )
+
+
+@dataclass
+class ServicePackageRoleDefault:
+    config: PackagesConfig = field(default_factory=PackagesConfig )
+    imports: list[str] = field(default_factory=list )
+    machines: list[str] = field(default_factory=list )
+    tags: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServicePackageRole:
+    default: ServicePackageRoleDefault
+
+
+@dataclass
+class ServicePackage:
+    meta: ServicePackageMeta
+    roles: ServicePackageRole
+    config: PackagesConfig = field(default_factory=PackagesConfig )
+    machines: dict[str, ServicePackageMachine] = field(default_factory=dict )
+
+
+@dataclass
+class SingleDiskConfig:
+    device: str | None = field(default=None )
+
+
+@dataclass
+class ServiceSingleDiskMachine:
+    config: SingleDiskConfig = field(default_factory=SingleDiskConfig )
+    imports: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServiceSingleDiskMeta:
+    name: str
+    description: str | None = field(default=None )
+    icon: str | None = field(default=None )
+
+
+@dataclass
+class ServiceSingleDiskRoleDefault:
+    config: SingleDiskConfig = field(default_factory=SingleDiskConfig )
+    imports: list[str] = field(default_factory=list )
+    machines: list[str] = field(default_factory=list )
+    tags: list[str] = field(default_factory=list )
+
+
+@dataclass
+class ServiceSingleDiskRole:
+    default: ServiceSingleDiskRoleDefault
+
+
+@dataclass
+class ServiceSingleDisk:
+    meta: ServiceSingleDiskMeta
+    roles: ServiceSingleDiskRole
+    config: SingleDiskConfig = field(default_factory=SingleDiskConfig )
+    machines: dict[str, ServiceSingleDiskMachine] = field(default_factory=dict )
+
+
+@dataclass
+class Service:
+    borgbackup: dict[str, ServiceBorgbackup] = field(default_factory=dict )
+    packages: dict[str, ServicePackage] = field(default_factory=dict )
+    single_disk: dict[str, ServiceSingleDisk] = field(default_factory=dict , metadata={"original_name": "single-disk"})
+
+
+@dataclass
+class Inventory:
+    meta: Meta
+    services: Service
+    machines: dict[str, Machine] = field(default_factory=dict )

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

@@ -7,7 +7,7 @@ from ..api import API
 from ..clan_uri import FlakeId
 from ..errors import ClanError
 from ..git import commit_file
-from ..inventory import Inventory, Machine
+from ..inventory import Machine, MachineDeploy, get_path, load_inventory, save_inventory
 
 log = logging.getLogger(__name__)
 
@@ -20,11 +20,11 @@ def create_machine(flake: FlakeId, machine: Machine) -> None:
             "Machine name must be a valid hostname", location="Create Machine"
         )
 
-    inventory = Inventory.load_file(flake.path)
+    inventory = load_inventory(flake.path)
     inventory.machines.update({machine.name: machine})
-    inventory.persist(flake.path, f"Create machine {machine.name}")
+    save_inventory(inventory, flake.path, f"Create machine {machine.name}")
 
-    commit_file(Inventory.get_path(flake.path), Path(flake.path))
+    commit_file(get_path(flake.path), Path(flake.path))
 
 
 def create_command(args: argparse.Namespace) -> None:
@@ -36,6 +36,7 @@ def create_command(args: argparse.Namespace) -> None:
             description=args.description,
             tags=args.tags,
             icon=args.icon,
+            deploy=MachineDeploy(),
         ),
     )
 

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

@@ -6,18 +6,18 @@ from ..clan_uri import FlakeId
 from ..completions import add_dynamic_completer, complete_machines
 from ..dirs import specific_machine_dir
 from ..errors import ClanError
-from ..inventory import Inventory
+from ..inventory import load_inventory, save_inventory
 
 
 @API.register
 def delete_machine(flake: FlakeId, name: str) -> None:
-    inventory = Inventory.load_file(flake.path)
+    inventory = load_inventory(flake.path)
 
     machine = inventory.machines.pop(name, None)
     if machine is None:
         raise ClanError(f"Machine {name} does not exist")
 
-    inventory.persist(flake.path, f"Delete machine {name}")
+    save_inventory(inventory, flake.path, f"Delete machine {name}")
 
     folder = specific_machine_dir(flake.path, name)
     if folder.exists():

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

@@ -4,7 +4,7 @@ import logging
 from pathlib import Path
 
 from clan_cli.api import API
-from clan_cli.inventory import Machine
+from clan_cli.inventory import Machine, from_dict
 
 from ..cmd import run_no_stdout
 from ..nix import nix_eval
@@ -24,7 +24,7 @@ def list_machines(flake_url: str | Path, debug: bool = False) -> dict[str, Machi
     proc = run_no_stdout(cmd)
 
     res = proc.stdout.strip()
-    data = {name: Machine.from_dict(v) for name, v in json.loads(res).items()}
+    data = {name: from_dict(Machine, v) for name, v in json.loads(res).items()}
     return data
 
 

pkgs/clan-cli/default.nix

@@ -21,6 +21,9 @@
   clan-core-path,
   nixpkgs,
   includedRuntimeDeps,
+
+  inventory-schema,
+  classgen,
 }:
 let
   pythonDependencies = [
@@ -60,6 +63,8 @@ let
     rm $out/clan_cli/config/jsonschema
     ln -sf ${nixpkgs'} $out/clan_cli/nixpkgs
     cp -r ${../../lib/jsonschema} $out/clan_cli/config/jsonschema
+
+    ${classgen}/bin/classgen ${inventory-schema}/schema.json $out/clan_cli/inventory/classes.py
   '';
 
   # Create a custom nixpkgs for use within the project

pkgs/clan-cli/flake-module.nix

@@ -9,7 +9,7 @@
     { self', pkgs, ... }:
     let
       flakeLock = lib.importJSON (self + /flake.lock);
-      flakeInputs = (builtins.removeAttrs inputs [ "self" ]);
+      flakeInputs = builtins.removeAttrs inputs [ "self" ];
       flakeLockVendoredDeps = flakeLock // {
         nodes =
           flakeLock.nodes
@@ -38,7 +38,6 @@
           '';
     in
     {
-
       devShells.clan-cli = pkgs.callPackage ./shell.nix {
         inherit (self'.packages) clan-cli clan-cli-full;
         inherit self';
@@ -46,6 +45,7 @@
       packages = {
         clan-cli = pkgs.python3.pkgs.callPackage ./default.nix {
           inherit (inputs) nixpkgs;
+          inherit (self'.packages) inventory-schema classgen;
           clan-core-path = clanCoreWithVendoredDeps;
           includedRuntimeDeps = [
             "age"
@@ -54,6 +54,7 @@
         };
         clan-cli-full = pkgs.python3.pkgs.callPackage ./default.nix {
           inherit (inputs) nixpkgs;
+          inherit (self'.packages) inventory-schema classgen;
           clan-core-path = clanCoreWithVendoredDeps;
           includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-programs.json;
         };
@@ -64,6 +65,8 @@
           buildInputs = [ pkgs.python3 ];
 
           installPhase = ''
+            ${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json ./clan_cli/inventory/classes.py
+
             python docs.py reference
             mkdir -p $out
             cp -r out/* $out
@@ -77,6 +80,8 @@
           buildInputs = [ pkgs.python3 ];
 
           installPhase = ''
+            ${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json ./clan_cli/inventory/classes.py
+
             python api.py > $out
           '';
         };
@@ -84,6 +89,35 @@
         default = self'.packages.clan-cli;
       };
 
-      checks = self'.packages.clan-cli.tests;
+      checks = self'.packages.clan-cli.tests // {
+        inventory-classes-up-to-date = pkgs.stdenv.mkDerivation {
+          name = "inventory-classes-up-to-date";
+          src = ./clan_cli/inventory;
+
+          env = {
+            classFile = "classes.py";
+          };
+          installPhase = ''
+            ${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json b_classes.py
+            file1=$classFile
+            file2=b_classes.py
+
+            echo "Comparing $file1 and $file2"
+            if cmp -s "$file1" "$file2"; then
+                echo "Files are identical"
+                echo "Classes file is up to date"
+            else
+                echo "Classes file is out of date or has been modified"
+                echo "run ./update.sh in the inventory directory to update the classes file"
+                echo "--------------------------------\n"
+                diff "$file1" "$file2"
+                echo "--------------------------------\n\n"
+                exit 1
+            fi
+
+            touch $out
+          '';
+        };
+      };
     };
 }

pkgs/clan-cli/shell.nix

@@ -45,5 +45,9 @@ mkShell {
 
     # Needed for impure tests
     ln -sfT ${clan-cli.nixpkgs} "$PKG_ROOT/clan_cli/nixpkgs"
+
+    # Generate classes.py from inventory schema
+    # This file is in .gitignore
+    ${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json $PKG_ROOT/clan_cli/inventory/classes.py
   '';
 }

pkgs/clan-cli/tests/test_machines_config.py

@@ -8,7 +8,7 @@ from clan_cli.config.machine import (
     verify_machine_config,
 )
 from clan_cli.config.schema import machine_schema
-from clan_cli.inventory import Machine
+from clan_cli.inventory import Machine, MachineDeploy
 from clan_cli.machines.create import create_machine
 from clan_cli.machines.list import list_machines
 
@@ -31,6 +31,7 @@ def test_create_machine_on_minimal_clan(test_flake_minimal: FlakeForTest) -> Non
             description="A test machine",
             tags=["test"],
             icon=None,
+            deploy=MachineDeploy(),
         ),
     )
 

pkgs/clan-cli/tests/test_modules.py

@@ -4,9 +4,19 @@ from typing import TYPE_CHECKING
 import pytest
 from fixtures_flakes import FlakeForTest
 
-from clan_cli.api.modules import list_modules, update_module_instance
+from clan_cli.api.modules import list_modules
 from clan_cli.clan_uri import FlakeId
-from clan_cli.inventory import Machine, Role, Service, ServiceMeta
+from clan_cli.inventory import (
+    Machine,
+    MachineDeploy,
+    ServiceBorgbackup,
+    ServiceBorgbackupMeta,
+    ServiceBorgbackupRole,
+    ServiceBorgbackupRoleClient,
+    ServiceBorgbackupRoleServer,
+    load_inventory,
+    save_inventory,
+)
 from clan_cli.machines.create import create_machine
 from clan_cli.nix import nix_eval, run_no_stdout
 
@@ -51,21 +61,30 @@ def test_add_module_to_inventory(
         ]
     )
     create_machine(
-        FlakeId(base_path), Machine(name="machine1", tags=[], system="x86_64-linux")
-    )
-    update_module_instance(
-        base_path,
-        "borgbackup",
-        "borgbackup1",
-        Service(
-            meta=ServiceMeta(name="borgbackup"),
-            roles={
-                "client": Role(machines=["machine1"]),
-                "server": Role(machines=["machine1"]),
-            },
+        FlakeId(base_path),
+        Machine(
+            name="machine1", tags=[], system="x86_64-linux", deploy=MachineDeploy()
         ),
     )
 
+    inventory = load_inventory(base_path)
+
+    inventory.services.borgbackup = {
+        "borg1": ServiceBorgbackup(
+            meta=ServiceBorgbackupMeta(name="borg1"),
+            roles=ServiceBorgbackupRole(
+                client=ServiceBorgbackupRoleClient(
+                    machines=["machine1"],
+                ),
+                server=ServiceBorgbackupRoleServer(
+                    machines=["machine1"],
+                ),
+            ),
+        )
+    }
+
+    save_inventory(inventory, base_path, "Add borgbackup service")
+
     cmd = ["facts", "generate", "--flake", str(test_flake_with_core.path), "machine1"]
     cli.run(cmd)
 

pkgs/classgen/main.py

@@ -1,28 +1,36 @@
+# ruff: noqa: RUF001
 import argparse
 import json
 from typing import Any
 
 
-# Function to map JSON schema types to Python types
-def map_json_type(json_type: Any, nested_type: str = "Any") -> str:
+# Function to map JSON schemas and types to Python types
+def map_json_type(
+    json_type: Any, nested_types: set[str] = {"Any"}, parent: Any = None
+) -> set[str]:
     if isinstance(json_type, list):
-        return " | ".join(map(map_json_type, json_type))
+        res = set()
+        for t in json_type:
+            res |= map_json_type(t)
+        return res
     if isinstance(json_type, dict):
         return map_json_type(json_type.get("type"))
     elif json_type == "string":
-        return "str"
+        return {"str"}
     elif json_type == "integer":
-        return "int"
+        return {"int"}
     elif json_type == "boolean":
-        return "bool"
+        return {"bool"}
     elif json_type == "array":
-        return f"list[{nested_type}]"  # Further specification can be handled if needed
+        assert nested_types, f"Array type not found for {parent}"
+        return {f"""list[{" | ".join(nested_types)}]"""}
     elif json_type == "object":
-        return f"dict[str, {nested_type}]"
+        assert nested_types, f"dict type not found for {parent}"
+        return {f"""dict[str, {" | ".join(nested_types)}]"""}
     elif json_type == "null":
-        return "None"
+        return {"None"}
     else:
-        return "Any"
+        raise ValueError(f"Python type not found for {json_type}")
 
 
 known_classes = set()
@@ -32,9 +40,9 @@ root_class = "Inventory"
 # Recursive function to generate dataclasses from JSON schema
 def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) -> str:
     properties = schema.get("properties", {})
-    required = schema.get("required", [])
 
-    fields = []
+    required_fields = []
+    fields_with_default = []
     nested_classes = []
 
     for prop, prop_info in properties.items():
@@ -42,77 +50,120 @@ def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) ->
 
         prop_type = prop_info.get("type", None)
         union_variants = prop_info.get("oneOf", [])
+        # Collect all types
+        field_types = set()
 
         title = prop_info.get("title", prop.removesuffix("s"))
         title_sanitized = "".join([p.capitalize() for p in title.split("-")])
         nested_class_name = f"""{class_name if class_name != root_class and not prop_info.get("title") else ""}{title_sanitized}"""
 
-        # if nested_class_name == "ServiceBorgbackupRoleServerConfig":
-        #     breakpoint()
-
         if (prop_type is None) and (not union_variants):
             raise ValueError(f"Type not found for property {prop} {prop_info}")
 
-        # Unions fields (oneOf)
-        # str | int | None
-        python_type = None
-
         if union_variants:
-            python_type = map_json_type(union_variants)
+            field_types = map_json_type(union_variants)
+
         elif prop_type == "array":
             item_schema = prop_info.get("items")
+
             if isinstance(item_schema, dict):
-                python_type = map_json_type(
-                    prop_type,
-                    map_json_type(item_schema),
+                field_types = map_json_type(
+                    prop_type, map_json_type(item_schema), field_name
                 )
-        else:
-            python_type = map_json_type(
-                prop_type,
-                map_json_type([i for i in prop_info.get("items", [])]),
-            )
 
-        assert python_type, f"Python type not found for {prop} {prop_info}"
+        elif prop_type == "object":
+            inner_type = prop_info.get("additionalProperties")
+            if inner_type and inner_type.get("type") == "object":
+                # Inner type is a class
+                field_types = map_json_type(prop_type, {nested_class_name}, field_name)
 
-        if prop in required:
-            field_def = f"{prop}: {python_type}"
-        else:
-            field_def = f"{prop}: {python_type} | None = None"
+                #
+                if nested_class_name not in known_classes:
+                    nested_classes.append(
+                        generate_dataclass(inner_type, nested_class_name)
+                    )
+                    known_classes.add(nested_class_name)
 
-        if prop_type == "object":
-            map_type = prop_info.get("additionalProperties")
-            if map_type:
-                # breakpoint()
-                if map_type.get("type") == "object":
-                    # Non trivial type
-                    if nested_class_name not in known_classes:
-                        nested_classes.append(
-                            generate_dataclass(map_type, nested_class_name)
-                        )
-                        known_classes.add(nested_class_name)
-                    field_def = f"{field_name}: dict[str, {nested_class_name}]"
-                else:
-                    # Trivial type
-                    field_def = f"{field_name}: dict[str, {map_json_type(map_type)}]"
-            else:
+            elif inner_type and inner_type.get("type") != "object":
+                # Trivial type
+                field_types = map_json_type(inner_type)
+
+            elif not inner_type:
+                # The type is a class
+                field_types = {nested_class_name}
                 if nested_class_name not in known_classes:
                     nested_classes.append(
                         generate_dataclass(prop_info, nested_class_name)
                     )
                     known_classes.add(nested_class_name)
+        else:
+            field_types = map_json_type(
+                prop_type,
+                nested_types=set(),
+                parent=field_name,
+            )
 
-                field_def = f"{field_name}: {nested_class_name}"
+        assert field_types, f"Python type not found for {prop} {prop_info}"
 
-        elif prop_type == "array":
-            items = prop_info.get("items", {})
-            if items.get("type") == "object":
-                nested_class_name = prop.capitalize()
-                nested_classes.append(generate_dataclass(items, nested_class_name))
-                field_def = f"{field_name}: List[{nested_class_name}]"
+        serialised_types = " | ".join(field_types)
+        field_meta = None
+        if field_name != prop:
+            field_meta = f"""{{"original_name": "{prop}"}}"""
 
-        fields.append(field_def)
+        field_def = f"{field_name}: {serialised_types}"
+        if field_meta:
+            field_def = f"{field_def} = field(metadata={field_meta})"
 
-    fields_str = "\n    ".join(fields)
+        if "default" in prop_info or field_name not in prop_info.get("required", []):
+            if "default" in prop_info:
+                default_value = prop_info.get("default")
+                if default_value is None:
+                    field_types |= {"None"}
+                    serialised_types = " | ".join(field_types)
+
+                    field_def = f"""{field_name}: {serialised_types} = field(default=None {f", metadata={field_meta}" if field_meta else ""})"""
+                elif isinstance(default_value, list):
+                    field_def = f"""{field_def} = field(default_factory=list {f", metadata={field_meta}" if field_meta else ""})"""
+                elif isinstance(default_value, dict):
+                    serialised_types = " | ".join(field_types)
+                    if serialised_types == nested_class_name:
+                        field_def = f"""{field_name}: {serialised_types} = field(default_factory={nested_class_name} {f", metadata={field_meta}" if field_meta else ""})"""
+                    elif f"dict[str, {nested_class_name}]" in serialised_types:
+                        field_def = f"""{field_name}: {serialised_types} = field(default_factory=dict {f", metadata={field_meta}" if field_meta else ""})"""
+                    else:
+                        field_def = f"""{field_name}: {serialised_types} | dict[str,Any] = field(default_factory=dict {f", metadata={field_meta}" if field_meta else ""})"""
+                elif default_value == "‹name›":
+                    # Special case for nix submodules
+                    pass
+                elif isinstance(default_value, str):
+                    field_def = f"""{field_name}: {serialised_types} = field(default = '{default_value}' {f", metadata={field_meta}" if field_meta else ""})"""
+                else:
+                    # Other default values unhandled yet.
+                    raise ValueError(
+                        f"Unhandled default value for field '{field_name}' - default value: {default_value}"
+                    )
+
+                fields_with_default.append(field_def)
+
+            if "default" not in prop_info:
+                # Field is not required and but also specifies no default value
+                # Trying to infer default value from type
+                if "dict" in str(serialised_types):
+                    field_def = f"""{field_name}: {serialised_types} = field(default_factory=dict {f", metadata={field_meta}" if field_meta else ""})"""
+                    fields_with_default.append(field_def)
+                elif "list" in str(serialised_types):
+                    field_def = f"""{field_name}: {serialised_types} = field(default_factory=list {f", metadata={field_meta}" if field_meta else ""})"""
+                    fields_with_default.append(field_def)
+                elif "None" in str(serialised_types):
+                    field_def = f"""{field_name}: {serialised_types} = field(default=None {f", metadata={field_meta}" if field_meta else ""})"""
+                    fields_with_default.append(field_def)
+                else:
+                    # Field is not required and but also specifies no default value
+                    required_fields.append(field_def)
+        else:
+            required_fields.append(field_def)
+
+    fields_str = "\n    ".join(required_fields + fields_with_default)
     nested_classes_str = "\n\n".join(nested_classes)
 
     class_def = f"@dataclass\nclass {class_name}:\n    {fields_str}\n"
@@ -128,12 +179,13 @@ def run_gen(args: argparse.Namespace) -> None:
 
     with open(args.output, "w") as f:
         f.write(
-            """
-# DON NOT EDIT THIS FILE MANUALLY. IT IS GENERATED.
-# UPDATE:
+            """# DON NOT EDIT THIS FILE MANUALLY. IT IS GENERATED.
+#
 # ruff: noqa: N815
 # ruff: noqa: N806
-from dataclasses import dataclass
+# ruff: noqa: F401
+# fmt: off
+from dataclasses import dataclass, field
 from typing import Any\n\n
 """
         )