forked from clan/clan-core
Inventory persistence improves error resistance
This commit is contained in:
parent
9f484c1d39
commit
f7c80834cb
@ -1,5 +1,4 @@
|
||||
import json
|
||||
import re
|
||||
from dataclasses import asdict, dataclass, is_dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
@ -56,16 +55,6 @@ class Machine:
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict[str, Any]) -> "Machine":
|
||||
if "name" not in d:
|
||||
raise ClanError("name not found in machine")
|
||||
|
||||
hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
|
||||
if not re.match(hostname_regex, d["name"]):
|
||||
raise ClanError(
|
||||
"Machine name must be a valid hostname",
|
||||
description=f"""Machine name: {d["name"]}""",
|
||||
)
|
||||
|
||||
return Machine(**d)
|
||||
|
||||
|
||||
@ -95,15 +84,9 @@ class Service:
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict[str, Any]) -> "Service":
|
||||
if "meta" not in d:
|
||||
raise ClanError("meta not found in service")
|
||||
|
||||
if "roles" not in d:
|
||||
raise ClanError("roles not found in service")
|
||||
|
||||
return Service(
|
||||
meta=ServiceMeta(**d["meta"]),
|
||||
roles={name: Role(**role) for name, role in d["roles"].items()},
|
||||
meta=ServiceMeta(**d.get("meta", {})),
|
||||
roles={name: Role(**role) for name, role in d.get("roles", {}).items()},
|
||||
machines={
|
||||
name: MachineServiceConfig(**machine)
|
||||
for name, machine in d.get("machines", {}).items()
|
||||
@ -118,23 +101,17 @@ class Inventory:
|
||||
|
||||
@staticmethod
|
||||
def from_dict(d: dict[str, Any]) -> "Inventory":
|
||||
if "machines" not in d:
|
||||
raise ClanError("machines not found in inventory")
|
||||
|
||||
if "services" not in d:
|
||||
raise ClanError("services not found in inventory")
|
||||
|
||||
return Inventory(
|
||||
machines={
|
||||
name: Machine.from_dict(machine)
|
||||
for name, machine in d["machines"].items()
|
||||
for name, machine in d.get("machines", {}).items()
|
||||
},
|
||||
services={
|
||||
name: {
|
||||
role: Service.from_dict(service)
|
||||
for role, service in services.items()
|
||||
}
|
||||
for name, services in d["services"].items()
|
||||
for name, services in d.get("services", {}).items()
|
||||
},
|
||||
)
|
||||
|
||||
|
@ -15,13 +15,14 @@ log = logging.getLogger(__name__)
|
||||
def create_machine(flake_dir: str | Path, machine: Machine) -> None:
|
||||
hostname_regex = r"^(?!-)[A-Za-z0-9-]{1,63}(?<!-)$"
|
||||
if not re.match(hostname_regex, machine.name):
|
||||
raise ClanError("Machine name must be a valid hostname")
|
||||
raise ClanError(
|
||||
"Machine name must be a valid hostname", location="Create Machine"
|
||||
)
|
||||
|
||||
inventory = Inventory.load_file(flake_dir)
|
||||
inventory.machines.update({machine.name: machine})
|
||||
inventory.persist(flake_dir)
|
||||
|
||||
if flake_dir is not None:
|
||||
commit_file(Inventory.get_path(flake_dir), Path(flake_dir))
|
||||
|
||||
|
||||
|
Loading…
Reference in New Issue
Block a user