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

vars: move generator class and bound methods into seperate module

Browse Source
This commit is contained in:
Johannes Kirschbauer
2025-08-13 18:39:04 +02:00
parent 62ef90e959
commit 4312e3fc2f
12 changed files with 192 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkgs/clan-cli/clan_cli/vars/_types.py
View File

@@ -10,7 +10,7 @@ from clan_lib.flake import Flake
from clan_lib.ssh.host import Host
if TYPE_CHECKING:
from .generate import Generator, Var
from .generator import Generator, Var
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_cli/vars/check.py
View File

@@ -8,7 +8,7 @@ from clan_lib.flake import Flake, require_flake
from clan_lib.machines.machines import Machine
if TYPE_CHECKING:
from .generate import Var
from .generator import Var
log = logging.getLogger(__name__)

178
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -4,8 +4,6 @@ import os
import shutil
import sys
from contextlib import ExitStack
from dataclasses import dataclass, field
from functools import cached_property
from pathlib import Path
from tempfile import TemporaryDirectory
@@ -15,170 +13,23 @@ from clan_cli.completions import (
complete_services_for_machine,
)
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generator import Generator, GeneratorKey
from clan_cli.vars.migration import check_can_migrate, migrate_files
from clan_lib.api import API
from clan_lib.cmd import RunOpts, run
from clan_lib.errors import ClanError
from clan_lib.flake import Flake, require_flake
from clan_lib.flake import require_flake
from clan_lib.git import commit_files
from clan_lib.machines.list import list_full_machines
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config, nix_shell, nix_test_store
from .check import check_vars
from .graph import minimal_closure, requested_closure
from .prompt import Prompt, ask
from .var import Var
from .prompt import ask
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class GeneratorKey:
"""A key uniquely identifying a generator within a clan."""
machine: str | None
name: str
@dataclass
class Generator:
name: str
files: list[Var] = field(default_factory=list)
share: bool = False
prompts: list[Prompt] = field(default_factory=list)
dependencies: list[GeneratorKey] = field(default_factory=list)
migrate_fact: str | None = None
machine: str | None = None
_flake: "Flake | None" = None
@property
def key(self) -> GeneratorKey:
return GeneratorKey(machine=self.machine, name=self.name)
def __hash__(self) -> int:
return hash(self.key)
@cached_property
def exists(self) -> bool:
assert self.machine is not None
assert self._flake is not None
return check_vars(self.machine, self._flake, generator_name=self.name)
@classmethod
def get_machine_generators(
cls: type["Generator"],
machine_name: str,
flake: "Flake",
include_previous_values: bool = False,
) -> list["Generator"]:
"""
Get all generators for a machine from the flake.
Args:
machine_name (str): The name of the machine.
flake (Flake): The flake to get the generators from.
Returns:
list[Generator]: A list of (unsorted) generators for the machine.
"""
# Get all generator metadata in one select (safe fields only)
generators_data = flake.select_machine(
machine_name,
"config.clan.core.vars.generators.*.{share,dependencies,migrateFact,prompts}",
)
if not generators_data:
return []
# Get all file metadata in one select
files_data = flake.select_machine(
machine_name,
"config.clan.core.vars.generators.*.files.*.{secret,deploy,owner,group,mode,neededFor}",
)
from clan_lib.machines.machines import Machine
machine = Machine(name=machine_name, flake=flake)
pub_store = machine.public_vars_store
sec_store = machine.secret_vars_store
generators = []
for gen_name, gen_data in generators_data.items():
# Build files from the files_data
files = []
gen_files = files_data.get(gen_name, {})
for file_name, file_data in gen_files.items():
var = Var(
id=f"{gen_name}/{file_name}",
name=file_name,
secret=file_data["secret"],
deploy=file_data["deploy"],
owner=file_data["owner"],
group=file_data["group"],
mode=(
file_data["mode"]
if isinstance(file_data["mode"], int)
else int(file_data["mode"], 8)
),
needed_for=file_data["neededFor"],
_store=pub_store if not file_data["secret"] else sec_store,
)
files.append(var)
# Build prompts
prompts = [Prompt.from_nix(p) for p in gen_data.get("prompts", {}).values()]
generator = cls(
name=gen_name,
share=gen_data["share"],
files=files,
dependencies=[
GeneratorKey(machine=machine_name, name=dep)
for dep in gen_data["dependencies"]
],
migrate_fact=gen_data.get("migrateFact"),
prompts=prompts,
machine=machine_name,
_flake=flake,
)
generators.append(generator)
# TODO: This should be done in a non-mutable way.
if include_previous_values:
for generator in generators:
for prompt in generator.prompts:
prompt.previous_value = _get_previous_value(
machine, generator, prompt
)
return generators
def final_script(self) -> Path:
assert self.machine is not None
assert self._flake is not None
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
output = Path(
machine.select(
f'config.clan.core.vars.generators."{self.name}".finalScript'
)
)
if tmp_store := nix_test_store():
output = tmp_store.joinpath(*output.parts[1:])
return output
def validation(self) -> str | None:
assert self.machine is not None
assert self._flake is not None
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
return machine.select(
f'config.clan.core.vars.generators."{self.name}".validationHash'
)
def bubblewrap_cmd(generator: str, tmpdir: Path) -> list[str]:
test_store = nix_test_store()
@@ -406,23 +257,6 @@ def _ask_prompts(
return prompt_values
def _get_previous_value(
machine: "Machine",
generator: Generator,
prompt: Prompt,
) -> str | None:
if not prompt.persist:
return None
pub_store = machine.public_vars_store
if pub_store.exists(generator, prompt.name):
return pub_store.get(generator, prompt.name).decode()
sec_store = machine.secret_vars_store
if sec_store.exists(generator, prompt.name):
return sec_store.get(generator, prompt.name).decode()
return None
@API.register
def get_generators(
machine: Machine,
@@ -445,9 +279,7 @@ def get_generators(
from . import graph
vars_generators = Generator.get_machine_generators(machine.name, machine.flake)
generators: dict[GeneratorKey, Generator] = {
generator.key: generator for generator in vars_generators
}
generators = {generator.key: generator for generator in vars_generators}
result_closure = []
if generator_name is None: # all generators selected
@@ -466,7 +298,7 @@ def get_generators(
if include_previous_values:
for generator in result_closure:
for prompt in generator.prompts:
prompt.previous_value = _get_previous_value(machine, generator, prompt)
prompt.previous_value = generator.get_previous_value(machine, prompt)
return result_closure

176
pkgs/clan-cli/clan_cli/vars/generator.py Normal file
View File

@@ -0,0 +1,176 @@
import logging
from dataclasses import dataclass, field
from functools import cached_property
from pathlib import Path
from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_test_store
from .check import check_vars
from .prompt import Prompt
from .var import Var
log = logging.getLogger(__name__)
@dataclass(frozen=True)
class GeneratorKey:
"""A key uniquely identifying a generator within a clan."""
machine: str | None
name: str
@dataclass
class Generator:
name: str
files: list[Var] = field(default_factory=list)
share: bool = False
prompts: list[Prompt] = field(default_factory=list)
dependencies: list[GeneratorKey] = field(default_factory=list)
migrate_fact: str | None = None
machine: str | None = None
_flake: "Flake | None" = None
@property
def key(self) -> GeneratorKey:
return GeneratorKey(machine=self.machine, name=self.name)
def __hash__(self) -> int:
return hash(self.key)
@cached_property
def exists(self) -> bool:
assert self.machine is not None
assert self._flake is not None
return check_vars(self.machine, self._flake, generator_name=self.name)
@classmethod
def get_machine_generators(
cls: type["Generator"],
machine_name: str,
flake: "Flake",
include_previous_values: bool = False,
) -> list["Generator"]:
"""
Get all generators for a machine from the flake.
Args:
machine_name (str): The name of the machine.
flake (Flake): The flake to get the generators from.
Returns:
list[Generator]: A list of (unsorted) generators for the machine.
"""
# Get all generator metadata in one select (safe fields only)
generators_data = flake.select_machine(
machine_name,
"config.clan.core.vars.generators.*.{share,dependencies,migrateFact,prompts}",
)
if not generators_data:
return []
# Get all file metadata in one select
files_data = flake.select_machine(
machine_name,
"config.clan.core.vars.generators.*.files.*.{secret,deploy,owner,group,mode,neededFor}",
)
from clan_lib.machines.machines import Machine
machine = Machine(name=machine_name, flake=flake)
pub_store = machine.public_vars_store
sec_store = machine.secret_vars_store
generators = []
for gen_name, gen_data in generators_data.items():
# Build files from the files_data
files = []
gen_files = files_data.get(gen_name, {})
for file_name, file_data in gen_files.items():
var = Var(
id=f"{gen_name}/{file_name}",
name=file_name,
secret=file_data["secret"],
deploy=file_data["deploy"],
owner=file_data["owner"],
group=file_data["group"],
mode=(
file_data["mode"]
if isinstance(file_data["mode"], int)
else int(file_data["mode"], 8)
),
needed_for=file_data["neededFor"],
_store=pub_store if not file_data["secret"] else sec_store,
)
files.append(var)
# Build prompts
prompts = [Prompt.from_nix(p) for p in gen_data.get("prompts", {}).values()]
generator = cls(
name=gen_name,
share=gen_data["share"],
files=files,
dependencies=[
GeneratorKey(machine=machine_name, name=dep)
for dep in gen_data["dependencies"]
],
migrate_fact=gen_data.get("migrateFact"),
prompts=prompts,
machine=machine_name,
_flake=flake,
)
generators.append(generator)
# TODO: This should be done in a non-mutable way.
if include_previous_values:
for generator in generators:
for prompt in generator.prompts:
prompt.previous_value = generator.get_previous_value(
machine, prompt
)
return generators
def get_previous_value(
self,
machine: "Machine",
prompt: Prompt,
) -> str | None:
if not prompt.persist:
return None
pub_store = machine.public_vars_store
if pub_store.exists(self, prompt.name):
return pub_store.get(self, prompt.name).decode()
sec_store = machine.secret_vars_store
if sec_store.exists(self, prompt.name):
return sec_store.get(self, prompt.name).decode()
return None
def final_script(self) -> Path:
assert self.machine is not None
assert self._flake is not None
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
output = Path(
machine.select(
f'config.clan.core.vars.generators."{self.name}".finalScript'
)
)
if tmp_store := nix_test_store():
output = tmp_store.joinpath(*output.parts[1:])
return output
def validation(self) -> str | None:
assert self.machine is not None
assert self._flake is not None
from clan_lib.machines.machines import Machine
machine = Machine(name=self.machine, flake=self._flake)
return machine.select(
f'config.clan.core.vars.generators."{self.name}".validationHash'
)

2
pkgs/clan-cli/clan_cli/vars/get.py
View File

@@ -10,7 +10,7 @@ from clan_cli.completions import (
from clan_lib.errors import ClanError
from clan_lib.flake import Flake, require_flake
from .generate import Var
from .generator import Var
from .list import get_machine_vars
log = logging.getLogger(__name__)

3
pkgs/clan-cli/clan_cli/vars/list.py
View File

@@ -5,7 +5,8 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_lib.flake import Flake, require_flake
from clan_lib.machines.machines import Machine
from .generate import Var, get_generators
from .generate import get_generators
from .generator import Var
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_cli/vars/public_modules/in_repo.py
View File

@@ -3,7 +3,7 @@ from collections.abc import Iterable
from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_cli.vars.generator import Generator, Var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host

2
pkgs/clan-cli/clan_cli/vars/public_modules/vm.py
View File

@@ -4,7 +4,7 @@ from collections.abc import Iterable
from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_cli.vars.generator import Generator, Var
from clan_lib.dirs import vm_state_dir
from clan_lib.errors import ClanError
from clan_lib.flake import Flake

2
pkgs/clan-cli/clan_cli/vars/secret_modules/fs.py
View File

@@ -3,7 +3,7 @@ import tempfile
from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_cli.vars.generator import Generator, Var
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host

2
pkgs/clan-cli/clan_cli/vars/secret_modules/password_store.py
View File

@@ -7,7 +7,7 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_cli.vars.generator import Generator, Var
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host
from clan_lib.ssh.upload import upload

2
pkgs/clan-cli/clan_cli/vars/secret_modules/vm.py
View File

@@ -3,7 +3,7 @@ from collections.abc import Iterable
from pathlib import Path
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator, Var
from clan_cli.vars.generator import Generator, Var
from clan_lib.dirs import vm_state_dir
from clan_lib.flake import Flake
from clan_lib.ssh.host import Host

2
pkgs/clan-cli/clan_cli/vars/set.py
View File

@@ -13,7 +13,7 @@ from clan_lib.flake import Flake
from clan_lib.git import commit_files
from clan_lib.machines.machines import Machine
from .generate import Var
from .generator import Var
from .prompt import ask
log = logging.getLogger(__name__)