yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Classgen: add mapped keys and more stuff

Browse Source
This commit is contained in:
Johannes Kirschbauer
2024-07-18 19:18:58 +02:00
parent 7e84eaa4b3
commit 07965598f5
17 changed files with 178 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -9,11 +9,13 @@ democlan
example_clan example_clan
result* result*
/pkgs/clan-cli/clan_cli/nixpkgs /pkgs/clan-cli/clan_cli/nixpkgs
/pkgs/clan-cli/clan_cli/inventory/classes.py
/pkgs/clan-cli/clan_cli/webui/assets /pkgs/clan-cli/clan_cli/webui/assets
nixos.qcow2 nixos.qcow2
**/*.glade~ **/*.glade~
/docs/out /docs/out
# dream2nix # dream2nix
.dream2nix .dream2nix

3
clanModules/single-disk/default.nix
View File

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

22
pkgs/clan-cli/clan_cli/api/modules.py
View File

@@ -6,7 +6,7 @@ from pathlib import Path
from clan_cli.cmd import run_no_stdout from clan_cli.cmd import run_no_stdout
from clan_cli.errors import ClanCmdError, ClanError 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 clan_cli.nix import nix_eval
from . import API 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 @API.register
def get_inventory(base_path: str) -> Inventory: def get_inventory(base_path: str) -> Inventory:
return Inventory.load_file(base_path) return load_inventory(base_path)

10
pkgs/clan-cli/clan_cli/clan/create.py
View File

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

6
pkgs/clan-cli/clan_cli/clan/show.py
View File

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

10
pkgs/clan-cli/clan_cli/clan/update.py
View File

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

110
pkgs/clan-cli/clan_cli/inventory/__init__.py
View File

@@ -1,6 +1,6 @@
import dataclasses import dataclasses
import json import json
from dataclasses import asdict, fields, is_dataclass from dataclasses import fields, is_dataclass
from pathlib import Path from pathlib import Path
from types import UnionType from types import UnionType
from typing import Any, get_args, get_origin from typing import Any, get_args, get_origin
@@ -8,11 +8,33 @@ from typing import Any, get_args, get_origin
from clan_cli.errors import ClanError from clan_cli.errors import ClanError
from clan_cli.git import commit_file from clan_cli.git import commit_file
from .classes import Inventory as NixInventory from .classes import (
from .classes import Machine, Service Inventory,
from .classes import Meta as InventoryMeta Machine,
MachineDeploy,
Meta,
Service,
ServiceBorgbackup,
ServiceBorgbackupMeta,
ServiceBorgbackupRole,
ServiceBorgbackupRoleClient,
ServiceBorgbackupRoleServer,
)
__all__ = ["Service", "Machine", "InventoryMeta"] # 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: def sanitize_string(s: str) -> str:
@@ -28,8 +50,11 @@ def dataclass_to_dict(obj: Any) -> Any:
""" """
if is_dataclass(obj): if is_dataclass(obj):
return { return {
sanitize_string(k): dataclass_to_dict(v) # Use either the original name or name
for k, v in asdict(obj).items() # type: ignore 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): elif isinstance(obj, list | tuple):
return [dataclass_to_dict(item) for item in obj] return [dataclass_to_dict(item) for item in obj]
@@ -77,12 +102,13 @@ def from_dict(t: type, data: dict[str, Any] | None) -> Any:
# Attempt to create an instance of the data_class # Attempt to create an instance of the data_class
field_values = {} field_values = {}
for field in fields(t): for field in fields(t):
field_value = data.get(field.name) original_name = field.metadata.get("original_name", field.name)
field_value = data.get(original_name)
field_type = get_inner_type(field.type) # type: ignore field_type = get_inner_type(field.type) # type: ignore
if field.name in data: if original_name in data:
# The field is present
# If the field is another dataclass, recursively instantiate it # If the field is another dataclass, recursively instantiate it
if is_dataclass(field_type): if is_dataclass(field_type):
field_value = from_dict(field_type, field_value) field_value = from_dict(field_type, field_value)
@@ -111,7 +137,7 @@ def from_dict(t: type, data: dict[str, Any] | None) -> Any:
# Fields with default value # Fields with default value
# a: Int = 1 # a: Int = 1
# b: list = Field(default_factory=list) # b: list = Field(default_factory=list)
if field.name in data or field_value is not None: if original_name in data or field_value is not None:
field_values[field.name] = field_value field_values[field.name] = field_value
else: else:
# Fields without default value # Fields without default value
@@ -121,54 +147,54 @@ def from_dict(t: type, data: dict[str, Any] | None) -> Any:
return t(**field_values) return t(**field_values)
except (TypeError, ValueError) as e: except (TypeError, ValueError) as e:
print(f"Failed to instantiate {t.__name__}: {e}") print(f"Failed to instantiate {t.__name__}: {e} {data}")
return None return None
# raise ClanError(f"Failed to instantiate {t.__name__}: {e}")
class Inventory:
nix_inventory: NixInventory
def __init__(self) -> None:
self.nix_inventory = NixInventory(
meta=InventoryMeta(name="New Clan"), machines={}, services=Service()
)
@staticmethod
def get_path(flake_dir: str | Path) -> Path: 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() return (Path(flake_dir) / "inventory.json").resolve()
@staticmethod
def load_file(flake_dir: str | Path) -> "Inventory": # Default inventory
inventory = from_dict( default_inventory = Inventory(
NixInventory, meta=Meta(name="New Clan"), machines={}, services=Service()
{
"meta": {"name": "New Clan"},
"machines": {},
"services": {},
},
) )
NixInventory(
meta=InventoryMeta(name="New Clan"), machines={}, services=Service()
)
inventory_file = Inventory.get_path(flake_dir) 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(): if inventory_file.exists():
with open(inventory_file) as f: with open(inventory_file) as f:
try: try:
res = json.load(f) res = json.load(f)
inventory = from_dict(NixInventory, res) inventory = from_dict(Inventory, res)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
# Error decoding the inventory file
raise ClanError(f"Error decoding inventory file: {e}") raise ClanError(f"Error decoding inventory file: {e}")
res = Inventory() return inventory
res.nix_inventory = inventory
return Inventory()
def persist(self, flake_dir: str | Path, message: str) -> None:
inventory_file = Inventory.get_path(flake_dir) 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)
with open(inventory_file, "w") as f: with open(inventory_file, "w") as f:
json.dump(dataclass_to_dict(self.nix_inventory), f, indent=2) json.dump(dataclass_to_dict(inventory), f, indent=2)
commit_file(inventory_file, Path(flake_dir), commit_message=message) commit_file(inventory_file, Path(flake_dir), commit_message=message)

175
pkgs/clan-cli/clan_cli/inventory/classes.py
View File

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

9
pkgs/clan-cli/clan_cli/machines/create.py
View File

@@ -7,7 +7,7 @@ from ..api import API
from ..clan_uri import FlakeId from ..clan_uri import FlakeId
from ..errors import ClanError from ..errors import ClanError
from ..git import commit_file 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__) 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" "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.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: def create_command(args: argparse.Namespace) -> None:
@@ -36,6 +36,7 @@ def create_command(args: argparse.Namespace) -> None:
description=args.description, description=args.description,
tags=args.tags, tags=args.tags,
icon=args.icon, icon=args.icon,
deploy=MachineDeploy(),
), ),
) )

6
pkgs/clan-cli/clan_cli/machines/delete.py
View File

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

4
pkgs/clan-cli/clan_cli/machines/list.py
View File

@@ -4,7 +4,7 @@ import logging
from pathlib import Path from pathlib import Path
from clan_cli.api import API 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 ..cmd import run_no_stdout
from ..nix import nix_eval 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) proc = run_no_stdout(cmd)
res = proc.stdout.strip() 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 return data

5
pkgs/clan-cli/default.nix
View File

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

37
pkgs/clan-cli/flake-module.nix
View File

@@ -45,6 +45,7 @@
packages = { packages = {
clan-cli = pkgs.python3.pkgs.callPackage ./default.nix { clan-cli = pkgs.python3.pkgs.callPackage ./default.nix {
inherit (inputs) nixpkgs; inherit (inputs) nixpkgs;
inherit (self'.packages) inventory-schema classgen;
clan-core-path = clanCoreWithVendoredDeps; clan-core-path = clanCoreWithVendoredDeps;
includedRuntimeDeps = [ includedRuntimeDeps = [
"age" "age"
@@ -53,6 +54,7 @@
}; };
clan-cli-full = pkgs.python3.pkgs.callPackage ./default.nix { clan-cli-full = pkgs.python3.pkgs.callPackage ./default.nix {
inherit (inputs) nixpkgs; inherit (inputs) nixpkgs;
inherit (self'.packages) inventory-schema classgen;
clan-core-path = clanCoreWithVendoredDeps; clan-core-path = clanCoreWithVendoredDeps;
includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-programs.json; includedRuntimeDeps = lib.importJSON ./clan_cli/nix/allowed-programs.json;
}; };
@@ -63,6 +65,8 @@
buildInputs = [ pkgs.python3 ]; buildInputs = [ pkgs.python3 ];
installPhase = '' installPhase = ''
${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json ./clan_cli/inventory/classes.py
python docs.py reference python docs.py reference
mkdir -p $out mkdir -p $out
cp -r out/* $out cp -r out/* $out
@@ -76,6 +80,8 @@
buildInputs = [ pkgs.python3 ]; buildInputs = [ pkgs.python3 ];
installPhase = '' installPhase = ''
${self'.packages.classgen}/bin/classgen ${self'.packages.inventory-schema}/schema.json ./clan_cli/inventory/classes.py
python api.py > $out python api.py > $out
''; '';
}; };
@@ -83,35 +89,6 @@
default = self'.packages.clan-cli; 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-pretty}/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
'';
};
};
}; };
} }

4
pkgs/clan-cli/shell.nix
View File

@@ -45,5 +45,9 @@ mkShell {
# Needed for impure tests # Needed for impure tests
ln -sfT ${clan-cli.nixpkgs} "$PKG_ROOT/clan_cli/nixpkgs" 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
''; '';
} }

3
pkgs/clan-cli/tests/test_machines_config.py
View File

@@ -8,7 +8,7 @@ from clan_cli.config.machine import (
verify_machine_config, verify_machine_config,
) )
from clan_cli.config.schema import machine_schema 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.create import create_machine
from clan_cli.machines.list import list_machines 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", description="A test machine",
tags=["test"], tags=["test"],
icon=None, icon=None,
deploy=MachineDeploy(),
), ),
) )

47
pkgs/clan-cli/tests/test_modules.py
View File

@@ -4,9 +4,19 @@ from typing import TYPE_CHECKING
import pytest import pytest
from fixtures_flakes import FlakeForTest 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.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.machines.create import create_machine
from clan_cli.nix import nix_eval, run_no_stdout from clan_cli.nix import nix_eval, run_no_stdout
@@ -51,21 +61,30 @@ def test_add_module_to_inventory(
] ]
) )
create_machine( create_machine(
FlakeId(base_path), Machine(name="machine1", tags=[], system="x86_64-linux") FlakeId(base_path),
) Machine(
update_module_instance( name="machine1", tags=[], system="x86_64-linux", deploy=MachineDeploy()
base_path,
"borgbackup",
"borgbackup1",
Service(
meta=ServiceMeta(name="borgbackup"),
roles={
"client": Role(machines=["machine1"]),
"server": Role(machines=["machine1"]),
},
), ),
) )
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"] cmd = ["facts", "generate", "--flake", str(test_flake_with_core.path), "machine1"]
cli.run(cmd) cli.run(cmd)

28
pkgs/classgen/main.py
View File

@@ -1,3 +1,4 @@
# ruff: noqa: RUF001
import argparse import argparse
import json import json
from typing import Any from typing import Any
@@ -105,7 +106,13 @@ def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) ->
assert field_types, f"Python type not found for {prop} {prop_info}" assert field_types, f"Python type not found for {prop} {prop_info}"
serialised_types = " | ".join(field_types) serialised_types = " | ".join(field_types)
field_meta = None
if field_name != prop:
field_meta = f"""{{"original_name": "{prop}"}}"""
field_def = f"{field_name}: {serialised_types}" field_def = f"{field_name}: {serialised_types}"
if field_meta:
field_def = f"{field_def} = field(metadata={field_meta})"
if "default" in prop_info or field_name not in prop_info.get("required", []): if "default" in prop_info or field_name not in prop_info.get("required", []):
if "default" in prop_info: if "default" in prop_info:
@@ -113,16 +120,23 @@ def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) ->
if default_value is None: if default_value is None:
field_types |= {"None"} field_types |= {"None"}
serialised_types = " | ".join(field_types) serialised_types = " | ".join(field_types)
field_def = f"{field_name}: {serialised_types} = None"
field_def = f"""{field_name}: {serialised_types} = field(default=None {f", metadata={field_meta}" if field_meta else ""})"""
elif isinstance(default_value, list): elif isinstance(default_value, list):
field_def = f"{field_def} = field(default_factory=list)" field_def = f"""{field_def} = field(default_factory=list {f", metadata={field_meta}" if field_meta else ""})"""
elif isinstance(default_value, dict): elif isinstance(default_value, dict):
field_types |= {"dict[str,Any]"}
serialised_types = " | ".join(field_types) serialised_types = " | ".join(field_types)
field_def = f"{field_name}: {serialised_types} = field(default_factory=dict)" 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›": elif default_value == "‹name›":
# Special case for nix submodules # Special case for nix submodules
pass 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: else:
# Other default values unhandled yet. # Other default values unhandled yet.
raise ValueError( raise ValueError(
@@ -135,13 +149,13 @@ def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) ->
# Field is not required and but also specifies no default value # Field is not required and but also specifies no default value
# Trying to infer default value from type # Trying to infer default value from type
if "dict" in str(serialised_types): if "dict" in str(serialised_types):
field_def = f"{field_name}: {serialised_types} = field(default_factory=dict)" 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) fields_with_default.append(field_def)
elif "list" in str(serialised_types): elif "list" in str(serialised_types):
field_def = f"{field_name}: {serialised_types} = field(default_factory=list)" 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) fields_with_default.append(field_def)
elif "None" in str(serialised_types): elif "None" in str(serialised_types):
field_def = f"{field_name}: {serialised_types} = None" field_def = f"""{field_name}: {serialised_types} = field(default=None {f", metadata={field_meta}" if field_meta else ""})"""
fields_with_default.append(field_def) fields_with_default.append(field_def)
else: else:
# Field is not required and but also specifies no default value # Field is not required and but also specifies no default value