Merge pull request 'chore(nix_models): use exported clan models' (#3773) from flake-models into main

Reviewed-on: https://git.clan.lol/clan/clan-core/pulls/3773
This commit is contained in:
hsjobeki
2025-05-28 12:25:07 +00:00
20 changed files with 342 additions and 98 deletions

View File

@@ -14,6 +14,7 @@ in
_prefix = lib.mkOption { _prefix = lib.mkOption {
type = types.listOf types.str; type = types.listOf types.str;
internal = true; internal = true;
visible = false;
default = [ ]; default = [ ];
}; };
self = lib.mkOption { self = lib.mkOption {
@@ -79,6 +80,8 @@ in
}; };
modules = lib.mkOption { modules = lib.mkOption {
# Correct type would be `types.attrsOf types.deferredModule` but that allows for
# Merging and transforms the value, which add eval overhead.
type = types.attrsOf types.raw; type = types.attrsOf types.raw;
default = { }; default = { };
description = '' description = ''

View File

@@ -1,4 +1,9 @@
{ self, inputs, ... }: {
self,
inputs,
options,
...
}:
let let
inputOverrides = builtins.concatStringsSep " " ( inputOverrides = builtins.concatStringsSep " " (
builtins.map (input: " --override-input ${input} ${inputs.${input}}") (builtins.attrNames inputs) builtins.map (input: " --override-input ${input} ${inputs.${input}}") (builtins.attrNames inputs)
@@ -28,6 +33,7 @@ in
legacyPackages.schemas = ( legacyPackages.schemas = (
import ./schemas { import ./schemas {
flakeOptions = options;
inherit inherit
pkgs pkgs
self self

View File

@@ -2,6 +2,7 @@
self, self,
self', self',
pkgs, pkgs,
flakeOptions,
... ...
}: }:
let let
@@ -21,6 +22,8 @@ let
import ../build-inventory/interface.nix { inherit (self) clanLib; } import ../build-inventory/interface.nix { inherit (self) clanLib; }
); );
clanSchema = jsonLib.parseOptions (flakeOptions.clan.type.getSubOptions [ "clan" ]) { };
renderSchema = pkgs.writers.writePython3Bin "render-schema" { renderSchema = pkgs.writers.writePython3Bin "render-schema" {
flakeIgnore = [ flakeIgnore = [
"F401" "F401"
@@ -28,12 +31,12 @@ let
]; ];
} ./render_schema.py; } ./render_schema.py;
inventory-schema-abstract = pkgs.stdenv.mkDerivation { clan-schema-abstract = pkgs.stdenv.mkDerivation {
name = "inventory-schema-files"; name = "clan-schema-files";
buildInputs = [ pkgs.cue ]; buildInputs = [ pkgs.cue ];
src = ./.; src = ./.;
buildPhase = '' buildPhase = ''
export SCHEMA=${builtins.toFile "inventory-schema.json" (builtins.toJSON inventorySchema)} export SCHEMA=${builtins.toFile "clan-schema.json" (builtins.toJSON clanSchema)}
cp $SCHEMA schema.json cp $SCHEMA schema.json
# Also generate a CUE schema version that is derived from the JSON schema # Also generate a CUE schema version that is derived from the JSON schema
cue import -f -p compose -l '#Root:' schema.json cue import -f -p compose -l '#Root:' schema.json
@@ -45,11 +48,13 @@ let
in in
{ {
inherit inherit
flakeOptions
frontMatterSchema frontMatterSchema
clanSchema
inventorySchema inventorySchema
modulesSchema modulesSchema
renderSchema renderSchema
inventory-schema-abstract clan-schema-abstract
; ;
# Inventory schema, with the modules schema added per role # Inventory schema, with the modules schema added per role

View File

@@ -9,7 +9,7 @@ from clan_lib.cmd import run
from clan_lib.errors import ClanCmdError, ClanError from clan_lib.errors import ClanCmdError, ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.nix import nix_eval from clan_lib.nix import nix_eval
from clan_lib.nix_models.inventory import Meta from clan_lib.nix_models.clan import InventoryMeta as Meta
log = logging.getLogger(__name__) log = logging.getLogger(__name__)

View File

@@ -2,8 +2,8 @@ from dataclasses import dataclass
from clan_lib.api import API from clan_lib.api import API
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.nix_models.inventory import Inventory, Meta from clan_lib.nix_models.clan import InventoryMeta as Meta
from clan_lib.persist.inventory_store import InventoryStore from clan_lib.persist.inventory_store import InventorySnapshot, InventoryStore
from clan_lib.persist.util import set_value_by_path from clan_lib.persist.util import set_value_by_path
@@ -14,7 +14,7 @@ class UpdateOptions:
@API.register @API.register
def update_clan_meta(options: UpdateOptions) -> Inventory: def update_clan_meta(options: UpdateOptions) -> InventorySnapshot:
inventory_store = InventoryStore(options.flake) inventory_store = InventoryStore(options.flake)
inventory = inventory_store.read() inventory = inventory_store.read()
set_value_by_path(inventory, "meta", options.meta) set_value_by_path(inventory, "meta", options.meta)

View File

@@ -9,12 +9,8 @@ from clan_lib.dirs import get_clan_flake_toplevel_or_env
from clan_lib.errors import ClanError from clan_lib.errors import ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.git import commit_file from clan_lib.git import commit_file
from clan_lib.nix_models.inventory import ( from clan_lib.nix_models.clan import InventoryMachine
Machine as InventoryMachine, from clan_lib.nix_models.clan import InventoryMachineDeploy as MachineDeploy
)
from clan_lib.nix_models.inventory import (
MachineDeploy,
)
from clan_lib.persist.inventory_store import InventoryStore from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import set_value_by_path from clan_lib.persist.util import set_value_by_path
from clan_lib.templates import ( from clan_lib.templates import (

View File

@@ -11,7 +11,7 @@ from clan_lib.errors import ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.machines.actions import get_machine from clan_lib.machines.actions import get_machine
from clan_lib.machines.machines import Machine from clan_lib.machines.machines import Machine
from clan_lib.nix_models.inventory import Machine as InventoryMachine from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.persist.inventory_store import InventoryStore from clan_lib.persist.inventory_store import InventoryStore
from clan_cli.completions import add_dynamic_completer, complete_tags from clan_cli.completions import add_dynamic_completer, complete_tags

View File

@@ -13,7 +13,7 @@ from clan_lib.errors import ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_build, nix_command from clan_lib.nix import nix_build, nix_command
from clan_lib.nix_models.inventory import Machine as InventoryMachine from clan_lib.nix_models.clan import InventoryMachine
from clan_cli.machines.create import CreateOptions, create_machine from clan_cli.machines.create import CreateOptions, create_machine
from clan_cli.vars.generate import generate_vars from clan_cli.vars.generate import generate_vars

View File

@@ -1,4 +1,7 @@
from clan_lib.nix_models.inventory import Inventory, Machine, Meta, Service from clan_lib.nix_models.clan import Inventory
from clan_lib.nix_models.clan import InventoryMachine as Machine
from clan_lib.nix_models.clan import InventoryMeta as Meta
from clan_lib.nix_models.clan import InventoryService as Service
def test_make_meta_minimal() -> None: def test_make_meta_minimal() -> None:

View File

@@ -8,9 +8,11 @@ from clan_cli.tests.fixtures_flakes import FlakeForTest
from clan_lib.api.modules import list_modules from clan_lib.api.modules import list_modules
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.nix import nix_eval, run from clan_lib.nix import nix_eval, run
from clan_lib.nix_models.inventory import ( from clan_lib.nix_models.clan import (
Machine, InventoryMachine as Machine,
MachineDeploy, )
from clan_lib.nix_models.clan import (
InventoryMachineDeploy as MachineDeploy,
) )
from clan_lib.persist.inventory_store import InventoryStore from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import set_value_by_path from clan_lib.persist.util import set_value_by_path

View File

@@ -7,8 +7,7 @@ from clan_lib.cmd import CmdOut, RunOpts, run
from clan_lib.errors import ClanError from clan_lib.errors import ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.nix import nix_command, nix_metadata, nix_shell from clan_lib.nix import nix_command, nix_metadata, nix_shell
from clan_lib.nix_models.inventory import Inventory from clan_lib.persist.inventory_store import InventorySnapshot, InventoryStore
from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.templates import ( from clan_lib.templates import (
InputPrio, InputPrio,
TemplateName, TemplateName,
@@ -35,7 +34,7 @@ class CreateOptions:
src_flake: Flake | None = None src_flake: Flake | None = None
input_prio: InputPrio | None = None input_prio: InputPrio | None = None
setup_git: bool = True setup_git: bool = True
initial: Inventory | None = None initial: InventorySnapshot | None = None
update_clan: bool = True update_clan: bool = True

View File

@@ -13,12 +13,11 @@ Interacting with 'clan_lib.inventory' is NOT recommended and will be removed
from clan_lib.api import API from clan_lib.api import API
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.nix_models.inventory import Inventory from clan_lib.persist.inventory_store import InventorySnapshot, InventoryStore
from clan_lib.persist.inventory_store import InventoryStore
@API.register @API.register
def get_inventory(flake: Flake) -> Inventory: def get_inventory(flake: Flake) -> InventorySnapshot:
inventory_store = InventoryStore(flake) inventory_store = InventoryStore(flake)
inventory = inventory_store.read() inventory = inventory_store.read()
return inventory return inventory

View File

@@ -1,8 +1,8 @@
from clan_lib.api import API from clan_lib.api import API
from clan_lib.errors import ClanError from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine from clan_lib.machines.machines import Machine
from clan_lib.nix_models.inventory import ( from clan_lib.nix_models.clan import (
Machine as InventoryMachine, InventoryMachine,
) )
from clan_lib.persist.inventory_store import InventoryStore from clan_lib.persist.inventory_store import InventoryStore
from clan_lib.persist.util import set_value_by_path from clan_lib.persist.util import set_value_by_path

View File

@@ -0,0 +1,200 @@
# DO NOT EDIT THIS FILE MANUALLY. IT IS GENERATED.
# This file was generated by running `pkgs/clan-cli/clan_lib.inventory/update.sh`
#
# ruff: noqa: N815
# ruff: noqa: N806
# ruff: noqa: F401
# fmt: off
from typing import Any, Literal, NotRequired, TypedDict
# Mimic "unknown".
# 'Any' is unsafe because it allows any operations
# This forces the user to use type-narrowing or casting in the code
class Unknown:
pass
InventoryInstanceModuleNameType = str
InventoryInstanceModuleInputType = str
class InventoryInstanceModule(TypedDict):
name: str
input: NotRequired[InventoryInstanceModuleInputType]
InventoryInstanceRoleMachineSettingsType = Unknown
class InventoryInstanceRoleMachine(TypedDict):
settings: NotRequired[InventoryInstanceRoleMachineSettingsType]
class InventoryInstanceRoleTag(TypedDict):
pass
InventoryInstanceRoleMachinesType = dict[str, InventoryInstanceRoleMachine]
InventoryInstanceRoleSettingsType = Unknown
InventoryInstanceRoleTagsType = dict[str, InventoryInstanceRoleTag]
class InventoryInstanceRole(TypedDict):
machines: NotRequired[InventoryInstanceRoleMachinesType]
settings: NotRequired[InventoryInstanceRoleSettingsType]
tags: NotRequired[InventoryInstanceRoleTagsType]
InventoryInstanceModuleType = InventoryInstanceModule
InventoryInstanceRolesType = dict[str, InventoryInstanceRole]
class InventoryInstance(TypedDict):
module: NotRequired[InventoryInstanceModuleType]
roles: NotRequired[InventoryInstanceRolesType]
InventoryMachineDeployTargethostType = str | None
class InventoryMachineDeploy(TypedDict):
targetHost: NotRequired[InventoryMachineDeployTargethostType]
InventoryMachineDeployType = InventoryMachineDeploy
InventoryMachineDescriptionType = str | None
InventoryMachineIconType = str | None
InventoryMachineMachineclassType = Literal["nixos", "darwin"]
InventoryMachineNameType = str
InventoryMachineTagsType = list[str]
class InventoryMachine(TypedDict):
deploy: NotRequired[InventoryMachineDeployType]
description: NotRequired[InventoryMachineDescriptionType]
icon: NotRequired[InventoryMachineIconType]
machineClass: NotRequired[InventoryMachineMachineclassType]
name: NotRequired[InventoryMachineNameType]
tags: NotRequired[InventoryMachineTagsType]
InventoryMetaNameType = str
InventoryMetaDescriptionType = str | None
InventoryMetaIconType = str | None
class InventoryMeta(TypedDict):
name: str
description: NotRequired[InventoryMetaDescriptionType]
icon: NotRequired[InventoryMetaIconType]
class InventoryService(TypedDict):
pass
InventoryInstancesType = dict[str, InventoryInstance]
InventoryMachinesType = dict[str, InventoryMachine]
InventoryMetaType = InventoryMeta
InventoryModulesType = dict[str, dict[str, Any] | list[Any] | bool | float | int | str | None]
InventoryServicesType = dict[str, InventoryService]
InventoryTagsType = dict[str, list[str]]
class Inventory(TypedDict):
instances: NotRequired[InventoryInstancesType]
machines: NotRequired[InventoryMachinesType]
meta: NotRequired[InventoryMetaType]
modules: NotRequired[InventoryModulesType]
services: NotRequired[InventoryServicesType]
tags: NotRequired[InventoryTagsType]
OutputModuleformachineType = dict[str, Unknown]
class Output(TypedDict):
moduleForMachine: NotRequired[OutputModuleformachineType]
SecretAgePluginsType = list[str]
class SecretAge(TypedDict):
plugins: NotRequired[SecretAgePluginsType]
SecretAgeType = SecretAge
class Secret(TypedDict):
age: NotRequired[SecretAgeType]
TemplateClanPathType = str
TemplateClanDescriptionType = str
class TemplateClan(TypedDict):
path: str
description: NotRequired[TemplateClanDescriptionType]
TemplateDiskoPathType = str
TemplateDiskoDescriptionType = str
class TemplateDisko(TypedDict):
path: str
description: NotRequired[TemplateDiskoDescriptionType]
TemplateMachinePathType = str
TemplateMachineDescriptionType = str
class TemplateMachine(TypedDict):
path: str
description: NotRequired[TemplateMachineDescriptionType]
TemplateClanType = dict[str, TemplateClan]
TemplateDiskoType = dict[str, TemplateDisko]
TemplateMachineType = dict[str, TemplateMachine]
class Template(TypedDict):
clan: NotRequired[TemplateClanType]
disko: NotRequired[TemplateDiskoType]
machine: NotRequired[TemplateMachineType]
ClanDirectoryType = dict[str, Any] | list[Any] | bool | float | int | str
ClanInventoryType = Inventory
ClanMachinesType = dict[str, Unknown]
ClanMetaType = Unknown
ClanModulesType = dict[str, dict[str, Any] | list[Any] | bool | float | int | str | None]
ClanOutputsType = Output
ClanSecretsType = Secret
ClanSelfType = dict[str, Any] | list[Any] | bool | float | int | str
ClanSpecialargsType = dict[str, dict[str, Any] | list[Any] | bool | float | int | str | None]
ClanTemplatesType = Template
class Clan(TypedDict):
directory: NotRequired[ClanDirectoryType]
inventory: NotRequired[ClanInventoryType]
machines: NotRequired[ClanMachinesType]
meta: NotRequired[ClanMetaType]
modules: NotRequired[ClanModulesType]
outputs: NotRequired[ClanOutputsType]
secrets: NotRequired[ClanSecretsType]
self: NotRequired[ClanSelfType]
specialArgs: NotRequired[ClanSpecialargsType]
templates: NotRequired[ClanTemplatesType]

View File

@@ -2,7 +2,7 @@
set -euo pipefail set -euo pipefail
jsonSchema=$(nix build .#schemas.inventory-schema-abstract --print-out-paths)/schema.json clanSchema=$(nix build .#schemas.clan-schema-abstract --print-out-paths)/schema.json
SCRIPT_DIR=$(dirname "$0") SCRIPT_DIR=$(dirname "$0")
cd "$SCRIPT_DIR" cd "$SCRIPT_DIR"
nix run .#classgen -- "$jsonSchema" "./inventory.py" nix run .#classgen -- "$clanSchema" "./clan.py"

View File

@@ -1,11 +1,17 @@
import json import json
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any, Protocol from typing import Any, NotRequired, Protocol, TypedDict
from clan_lib.errors import ClanError from clan_lib.errors import ClanError
from clan_lib.git import commit_file from clan_lib.git import commit_file
from clan_lib.nix_models.inventory import Inventory from clan_lib.nix_models.clan import (
Inventory,
InventoryInstancesType,
InventoryMachinesType,
InventoryMetaType,
InventoryServicesType,
)
from .util import ( from .util import (
calc_patches, calc_patches,
@@ -75,8 +81,8 @@ def sanitize(data: Any, whitelist_paths: list[str], current_path: list[str]) ->
@dataclass @dataclass
class WriteInfo: class WriteInfo:
writeables: dict[str, set[str]] writeables: dict[str, set[str]]
data_eval: Inventory data_eval: "InventorySnapshot"
data_disk: Inventory data_disk: "InventorySnapshot"
class FlakeInterface(Protocol): class FlakeInterface(Protocol):
@@ -92,6 +98,19 @@ class FlakeInterface(Protocol):
def path(self) -> Path: ... def path(self) -> Path: ...
class InventorySnapshot(TypedDict):
"""
Restricted view of an Inventory.
It contains only the keys that are convertible to python types and can be serialized to JSON.
"""
machines: NotRequired[InventoryMachinesType]
instances: NotRequired[InventoryInstancesType]
meta: NotRequired[InventoryMetaType]
services: NotRequired[InventoryServicesType]
class InventoryStore: class InventoryStore:
def __init__( def __init__(
self, self,
@@ -117,10 +136,11 @@ class InventoryStore:
self._allowed_path_transforms = _allowed_path_transforms self._allowed_path_transforms = _allowed_path_transforms
if _keys is None: if _keys is None:
_keys = ["machines", "instances", "meta", "services"] _keys = list(InventorySnapshot.__annotations__.keys())
self._keys = _keys self._keys = _keys
def _load_merged_inventory(self) -> Inventory: def _load_merged_inventory(self) -> InventorySnapshot:
""" """
Loads the evaluated inventory. Loads the evaluated inventory.
After all merge operations with eventual nix code in buildClan. After all merge operations with eventual nix code in buildClan.
@@ -140,7 +160,7 @@ class InventoryStore:
return sanitized return sanitized
def _get_persisted(self) -> Inventory: def _get_persisted(self) -> InventorySnapshot:
""" """
Load the inventory FILE from the flake directory Load the inventory FILE from the flake directory
If no file is found, returns an empty dictionary If no file is found, returns an empty dictionary
@@ -189,8 +209,8 @@ class InventoryStore:
""" """
current_priority = self._get_inventory_current_priority() current_priority = self._get_inventory_current_priority()
data_eval: Inventory = self._load_merged_inventory() data_eval: InventorySnapshot = self._load_merged_inventory()
data_disk: Inventory = self._get_persisted() data_disk: InventorySnapshot = self._get_persisted()
writeables = determine_writeability( writeables = determine_writeability(
current_priority, dict(data_eval), dict(data_disk) current_priority, dict(data_eval), dict(data_disk)
@@ -198,7 +218,7 @@ class InventoryStore:
return WriteInfo(writeables, data_eval, data_disk) return WriteInfo(writeables, data_eval, data_disk)
def read(self) -> Inventory: def read(self) -> InventorySnapshot:
""" """
Accessor to the merged inventory Accessor to the merged inventory
@@ -226,7 +246,9 @@ class InventoryStore:
commit_message=f"Delete inventory keys {delete_set}", commit_message=f"Delete inventory keys {delete_set}",
) )
def write(self, update: Inventory, message: str, commit: bool = True) -> None: def write(
self, update: InventorySnapshot, message: str, commit: bool = True
) -> None:
""" """
Write the inventory to the flake directory Write the inventory to the flake directory
and commit it to git with the given message and commit it to git with the given message

View File

@@ -24,8 +24,8 @@ from clan_lib.errors import ClanError
from clan_lib.flake import Flake from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_command from clan_lib.nix import nix_command
from clan_lib.nix_models.inventory import Machine as InventoryMachine from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.nix_models.inventory import MachineDeploy from clan_lib.nix_models.clan import InventoryMachineDeploy as MachineDeploy
from clan_lib.ssh.remote import Remote from clan_lib.ssh.remote import Remote
log = logging.getLogger(__name__) log = logging.getLogger(__name__)

View File

@@ -106,10 +106,10 @@
src = ./clan_lib/nix_models; src = ./clan_lib/nix_models;
env = { env = {
classFile = "inventory.py"; classFile = "clan.py";
}; };
installPhase = '' installPhase = ''
${self'.packages.classgen}/bin/classgen ${self'.legacyPackages.schemas.inventory-schema-abstract}/schema.json b_classes.py ${self'.packages.classgen}/bin/classgen ${self'.legacyPackages.schemas.clan-schema-abstract}/schema.json b_classes.py
file1=$classFile file1=$classFile
file2=b_classes.py file2=b_classes.py

View File

@@ -46,8 +46,7 @@ mkShell {
# Add clan command to PATH # Add clan command to PATH
export PATH="$PKG_ROOT/bin":"$PATH" export PATH="$PKG_ROOT/bin":"$PATH"
# Generate classes.py from inventory schema # Generate classes.py from schemas
# This file is in .gitignore ${self'.packages.classgen}/bin/classgen ${self'.legacyPackages.schemas.clan-schema-abstract}/schema.json $PKG_ROOT/clan_lib/nix_models/clan.py
${self'.packages.classgen}/bin/classgen ${self'.legacyPackages.schemas.inventory-schema-abstract}/schema.json $PKG_ROOT/clan_lib/nix_models/inventory.py
''; '';
} }

View File

@@ -3,7 +3,7 @@ import argparse
import json import json
import logging import logging
import sys import sys
from collections.abc import Callable from collections.abc import Callable, Iterable
from functools import partial from functools import partial
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@@ -15,58 +15,70 @@ class Error(Exception):
pass pass
def sort_types(items: Iterable[str]) -> list[str]:
def sort_key(item: str) -> tuple[int, str]:
# Priority order: lower number = higher priority
if item.startswith(("dict", "list")):
return (0, item) # Highest priority, dicts and lists should be first
if item == "None":
return (2, item) # Lowest priority, None should be last
return (1, item) # Middle priority, sorted alphabetically
return sorted(items, key=sort_key)
# Function to map JSON schemas and types to Python types # Function to map JSON schemas and types to Python types
def map_json_type( def map_json_type(
json_type: Any, nested_types: set[str] | None = None, parent: Any = None json_type: Any, nested_types: list[str] | None = None, parent: Any = None
) -> set[str]: ) -> list[str]:
if nested_types is None: if nested_types is None:
nested_types = {"Any"} nested_types = ["Any"]
if isinstance(json_type, list): if isinstance(json_type, list):
res = set() res: list[str] = []
for t in json_type: for t in json_type:
res |= map_json_type(t) res.extend(map_json_type(t))
return res return sort_types(set(res))
if isinstance(json_type, dict): if isinstance(json_type, dict):
items = json_type.get("items") items = json_type.get("items")
if items: if items:
nested_types = map_json_type(items) nested_types = map_json_type(items)
return map_json_type(json_type.get("type"), nested_types)
if not json_type.get("type") and json_type.get("tsType") == "unknown":
return ["Unknown"]
return sort_types(map_json_type(json_type.get("type"), nested_types))
if json_type == "string": if json_type == "string":
return {"str"} return ["str"]
if json_type == "integer": if json_type == "integer":
return {"int"} return ["int"]
if json_type == "number": if json_type == "number":
return {"float"} return ["float"]
if json_type == "boolean": if json_type == "boolean":
return {"bool"} return ["bool"]
# In Python, "number" is analogous to the float type. # In Python, "number" is analogous to the float type.
# https://json-schema.org/understanding-json-schema/reference/numeric#number # https://json-schema.org/understanding-json-schema/reference/numeric#number
if json_type == "number": if json_type == "number":
return {"float"} return ["float"]
if json_type == "array": if json_type == "array":
assert nested_types, f"Array type not found for {parent}" assert nested_types, f"Array type not found for {parent}"
return {f"""list[{" | ".join(nested_types)}]"""} return [f"""list[{" | ".join(sort_types(nested_types))}]"""]
if json_type == "object": if json_type == "object":
assert nested_types, f"dict type not found for {parent}" assert nested_types, f"dict type not found for {parent}"
return {f"""dict[str, {" | ".join(nested_types)}]"""} return [f"""dict[str, {" | ".join(sort_types(nested_types))}]"""]
if json_type == "null": if json_type == "null":
return {"None"} return ["None"]
msg = f"Python type not found for {json_type}" msg = f"Python type not found for {json_type}"
raise Error(msg) raise Error(msg)
known_classes = set() known_classes = set()
root_class = "Inventory"
# TODO: make this configurable # TODO: make this configurable
# For now this only includes static top-level attributes of the inventory. root_class = "Clan"
attrs = ["machines", "meta", "services", "instances"]
static: dict[str, str] = {"Service": "dict[str, Any]"}
def field_def_from_default_type( def field_def_from_default_type(
field_name: str, field_name: str,
field_types: set[str], field_types: list[str],
class_name: str, class_name: str,
finalize_field: Callable[..., tuple[str, str]], finalize_field: Callable[..., tuple[str, str]],
) -> tuple[str, str] | None: ) -> tuple[str, str] | None:
@@ -128,7 +140,7 @@ def field_def_from_default_type(
def field_def_from_default_value( def field_def_from_default_value(
default_value: Any, default_value: Any,
field_name: str, field_name: str,
field_types: set[str], field_types: list[str],
nested_class_name: str, nested_class_name: str,
finalize_field: Callable[..., tuple[str, str]], finalize_field: Callable[..., tuple[str, str]],
) -> tuple[str, str] | None: ) -> tuple[str, str] | None:
@@ -141,7 +153,7 @@ def field_def_from_default_value(
) )
if default_value is None: if default_value is None:
return finalize_field( return finalize_field(
field_types=field_types | {"None"}, field_types=[*field_types, "None"],
default="None", default="None",
) )
if isinstance(default_value, list): if isinstance(default_value, list):
@@ -189,7 +201,7 @@ def field_def_from_default_value(
def get_field_def( def get_field_def(
field_name: str, field_name: str,
field_meta: str | None, field_meta: str | None,
field_types: set[str], field_types: list[str],
default: str | None = None, default: str | None = None,
default_factory: str | None = None, default_factory: str | None = None,
type_appendix: str = "", type_appendix: str = "",
@@ -197,10 +209,10 @@ def get_field_def(
if "None" in field_types or default or default_factory: if "None" in field_types or default or default_factory:
if "None" in field_types: if "None" in field_types:
field_types.remove("None") field_types.remove("None")
serialised_types = " | ".join(field_types) + type_appendix serialised_types = " | ".join(sort_types(field_types)) + type_appendix
serialised_types = f"{serialised_types}" serialised_types = f"{serialised_types}"
else: else:
serialised_types = " | ".join(field_types) + type_appendix serialised_types = " | ".join(sort_types(field_types)) + type_appendix
return (field_name, serialised_types) return (field_name, serialised_types)
@@ -217,26 +229,21 @@ def generate_dataclass(
fields_with_default: list[tuple[str, str]] = [] fields_with_default: list[tuple[str, str]] = []
nested_classes: list[str] = [] nested_classes: list[str] = []
# if We are at the top level, and the attribute name is in shallow
# return f"{class_name} = dict[str, Any]"
if class_name in static:
return f"{class_name} = {static[class_name]}"
for prop, prop_info in properties.items(): for prop, prop_info in properties.items():
# If we are at the top level, and the attribute name is not explicitly included we only do shallow # If we are at the top level, and the attribute name is not explicitly included we only do shallow
field_name = prop.replace("-", "_") field_name = prop.replace("-", "_")
if len(attr_path) == 0 and prop not in attrs: # if len(attr_path) == 0 and prop in shallow_attrs:
field_def = field_name, "dict[str, Any]" # field_def = field_name, "dict[str, Any]"
fields_with_default.append(field_def) # fields_with_default.append(field_def)
continue # continue
prop_type = prop_info.get("type", None) prop_type = prop_info.get("type", None)
union_variants = prop_info.get("oneOf", []) union_variants = prop_info.get("oneOf", [])
enum_variants = prop_info.get("enum", []) enum_variants = prop_info.get("enum", [])
# Collect all types # Collect all types
field_types = set() field_types: list[str] = []
title = prop_info.get("title", prop.removesuffix("s")) title = prop_info.get("title", prop.removesuffix("s"))
title_sanitized = "".join([p.capitalize() for p in title.split("-")]) title_sanitized = "".join([p.capitalize() for p in title.split("-")])
@@ -259,13 +266,13 @@ def generate_dataclass(
) )
elif enum := prop_info.get("enum"): elif enum := prop_info.get("enum"):
literals = ", ".join([f'"{string}"' for string in enum]) literals = ", ".join([f'"{string}"' for string in enum])
field_types = {f"""Literal[{literals}]"""} field_types = [f"""Literal[{literals}]"""]
elif prop_type == "object": elif prop_type == "object":
inner_type = prop_info.get("additionalProperties") inner_type = prop_info.get("additionalProperties")
if inner_type and inner_type.get("type") == "object": if inner_type and inner_type.get("type") == "object":
# Inner type is a class # Inner type is a class
field_types = map_json_type(prop_type, {nested_class_name}, field_name) field_types = map_json_type(prop_type, [nested_class_name], field_name)
if nested_class_name not in known_classes: if nested_class_name not in known_classes:
nested_classes.append( nested_classes.append(
@@ -278,13 +285,13 @@ def generate_dataclass(
elif inner_type and inner_type.get("type") != "object": elif inner_type and inner_type.get("type") != "object":
# Trivial type: # Trivial type:
# dict[str, inner_type] # dict[str, inner_type]
field_types = { field_types = [
f"""dict[str, {" | ".join(map_json_type(inner_type))}]""" f"""dict[str, {" | ".join(map_json_type(inner_type))}]"""
} ]
elif not inner_type: elif not inner_type:
# The type is a class # The type is a class
field_types = {nested_class_name} field_types = [nested_class_name]
if nested_class_name not in known_classes: if nested_class_name not in known_classes:
nested_classes.append( nested_classes.append(
generate_dataclass( generate_dataclass(
@@ -293,11 +300,11 @@ def generate_dataclass(
) )
known_classes.add(nested_class_name) known_classes.add(nested_class_name)
elif prop_type == "Unknown": elif prop_type == "Unknown":
field_types = {"Unknown"} field_types = ["Unknown"]
else: else:
field_types = map_json_type( field_types = map_json_type(
prop_type, prop_type,
nested_types=set(), nested_types=[],
parent=field_name, parent=field_name,
) )
@@ -309,6 +316,9 @@ def generate_dataclass(
finalize_field = partial(get_field_def, field_name, field_meta) finalize_field = partial(get_field_def, field_name, field_meta)
# Sort and remove potential duplicates
field_types_sorted = sort_types(set(field_types))
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 prop_info.get("type") == "object": if prop_info.get("type") == "object":
prop_info.update({"default": {}}) prop_info.update({"default": {}})
@@ -318,7 +328,7 @@ def generate_dataclass(
field_def = field_def_from_default_value( field_def = field_def_from_default_value(
default_value=default_value, default_value=default_value,
field_name=field_name, field_name=field_name,
field_types=field_types, field_types=field_types_sorted,
nested_class_name=nested_class_name, nested_class_name=nested_class_name,
finalize_field=finalize_field, finalize_field=finalize_field,
) )
@@ -328,7 +338,7 @@ def generate_dataclass(
if not field_def: if not field_def:
# Finalize without the default value # Finalize without the default value
field_def = finalize_field( field_def = finalize_field(
field_types=field_types, field_types=field_types_sorted,
) )
required_fields.append(field_def) required_fields.append(field_def)
@@ -337,7 +347,7 @@ def generate_dataclass(
# Trying to infer default value from type # Trying to infer default value from type
field_def = field_def_from_default_type( field_def = field_def_from_default_type(
field_name=field_name, field_name=field_name,
field_types=field_types, field_types=field_types_sorted,
class_name=class_name, class_name=class_name,
finalize_field=finalize_field, finalize_field=finalize_field,
) )
@@ -346,13 +356,13 @@ def generate_dataclass(
fields_with_default.append(field_def) fields_with_default.append(field_def)
if not field_def: if not field_def:
field_def = finalize_field( field_def = finalize_field(
field_types=field_types, field_types=field_types_sorted,
) )
required_fields.append(field_def) required_fields.append(field_def)
else: else:
field_def = finalize_field( field_def = finalize_field(
field_types=field_types, field_types=field_types_sorted,
) )
required_fields.append(field_def) required_fields.append(field_def)