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

vars: move generation functions to clan_lib

Browse Source
This commit is contained in:
DavHau
2025-08-19 18:05:12 +07:00
parent 9ce563aa08
commit 815c6c9438
19 changed files with 189 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pkgs/clan-cli/clan_cli/tests/test_vars.py
View File

@@ -8,10 +8,6 @@ from clan_cli.tests.age_keys import SopsSetup
from clan_cli.tests.fixtures_flakes import ClanFlake
from clan_cli.tests.helpers import cli
from clan_cli.vars.check import check_vars
from clan_cli.vars.generate import (
get_generators,
run_generators,
)
from clan_cli.vars.generator import (
Generator,
GeneratorKey,
@@ -26,6 +22,10 @@ from clan_lib.errors import ClanError
from clan_lib.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_eval, run
from clan_lib.vars.generate import (
get_generators,
run_generators,
)
def test_dependencies_as_files(temp_dir: Path) -> None:

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

@@ -36,7 +36,7 @@ def vars_status(
# signals if a var needs to be updated (eg. needs re-encryption due to new users added)
unfixed_secret_vars = []
invalid_generators = []
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators(machine.name, machine.flake)
if generator_name:

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

@@ -10,7 +10,7 @@ log = logging.getLogger(__name__)
def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators(machine.name, machine.flake)
if generator_name:

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

@@ -1,175 +1,15 @@
import argparse
import logging
from collections.abc import Callable
from clan_cli.completions import (
add_dynamic_completer,
complete_machines,
complete_services_for_machine,
)
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.errors import ClanError
from clan_lib.flake import require_flake
from clan_lib.machines.list import list_full_machines
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config
from .graph import minimal_closure, requested_closure
log = logging.getLogger(__name__)
@API.register
def get_generators(
machine: Machine,
full_closure: bool,
generator_name: str | None = None,
include_previous_values: bool = False,
) -> list[Generator]:
"""
Get generators for a machine, with optional closure computation.
Args:
machine: The machine to get generators for.
full_closure: If True, include all dependency generators. If False, only include missing ones.
generator_name: Name of a specific generator to get, or None for all generators.
include_previous_values: If True, populate prompts with their previous values.
Returns:
List of generators based on the specified selection and closure mode.
"""
from . import graph
vars_generators = Generator.get_machine_generators(machine.name, machine.flake)
generators = {generator.key: generator for generator in vars_generators}
result_closure = []
if generator_name is None: # all generators selected
if full_closure:
result_closure = graph.full_closure(generators)
else:
result_closure = graph.all_missing_closure(generators)
# specific generator selected
elif full_closure:
gen_key = GeneratorKey(machine=machine.name, name=generator_name)
result_closure = requested_closure([gen_key], generators)
else:
gen_key = GeneratorKey(machine=machine.name, name=generator_name)
result_closure = minimal_closure([gen_key], generators)
if include_previous_values:
for generator in result_closure:
for prompt in generator.prompts:
prompt.previous_value = generator.get_previous_value(machine, prompt)
return result_closure
def _ensure_healthy(
machine: "Machine",
generators: list[Generator] | None = None,
) -> None:
"""
Run health checks on the provided generators.
Fails if any of the generators' health checks fail.
"""
if generators is None:
generators = Generator.get_machine_generators(machine.name, machine.flake)
pub_healtcheck_msg = machine.public_vars_store.health_check(
machine.name, generators
)
sec_healtcheck_msg = machine.secret_vars_store.health_check(
machine.name, generators
)
if pub_healtcheck_msg or sec_healtcheck_msg:
msg = f"Health check failed for machine {machine.name}:\n"
if pub_healtcheck_msg:
msg += f"Public vars store: {pub_healtcheck_msg}\n"
if sec_healtcheck_msg:
msg += f"Secret vars store: {sec_healtcheck_msg}"
raise ClanError(msg)
PromptFunc = Callable[[Generator], dict[str, str]]
"""Type for a function that collects prompt values for a generator.
The function receives a Generator and should return a dictionary mapping
prompt names to their values. This allows for custom prompt collection
strategies (e.g., interactive CLI, GUI, or programmatic).
"""
@API.register
def run_generators(
machines: list[Machine],
generators: str | list[str] = "minimal",
prompt_values: dict[str, dict[str, str]] | PromptFunc = lambda g: g.ask_prompts(),
no_sandbox: bool = False,
) -> None:
"""Run the specified generators for machines.
Args:
machines: The machines to run generators for.
generators: Can be:
- str: Single generator name to run (ensuring dependencies are met)
- list[str]: Specific generator names to run exactly as provided.
Dependency generators are not added automatically in this case.
The caller must ensure that all dependencies are included.
- "all": Run all generators (full closure)
- "minimal": Run only missing generators (minimal closure) (default)
prompt_values: A dictionary mapping generator names to their prompt values,
or a function that returns prompt values for a generator.
no_sandbox: Whether to disable sandboxing when executing the generator.
Raises:
ClanError: If the machine or generator is not found, or if there are issues with
executing the generator.
"""
for machine in machines:
if generators == "all":
generator_objects = get_generators(machine, full_closure=True)
elif generators == "minimal":
generator_objects = get_generators(machine, full_closure=False)
elif isinstance(generators, str) and generators not in ["all", "minimal"]:
# Single generator name - compute minimal closure for it
generator_objects = get_generators(
machine, full_closure=False, generator_name=generators
)
elif isinstance(generators, list):
if len(generators) == 0:
return
# Create GeneratorKeys for this specific machine
generator_keys = {
GeneratorKey(machine=machine.name, name=name) for name in generators
}
all_generators = get_generators(machine, full_closure=True)
generator_objects = [g for g in all_generators if g.key in generator_keys]
else:
msg = f"Invalid generators argument: {generators}. Must be 'all', 'minimal', a generator name, or a list of generator names"
raise ValueError(msg)
# If prompt function provided, ask all prompts
# TODO: make this more lazy and ask for every generator on execution
if callable(prompt_values):
prompt_values = {
generator.name: prompt_values(generator)
for generator in generator_objects
}
# execute health check
_ensure_healthy(machine=machine, generators=generator_objects)
# execute generators
for generator in generator_objects:
if check_can_migrate(machine, generator):
migrate_files(machine, generator)
else:
generator.execute(
machine=machine,
prompt_values=prompt_values.get(generator.name, {}),
no_sandbox=no_sandbox,
)
from clan_lib.vars.generate import run_generators
def generate_command(args: argparse.Namespace) -> None:

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

@@ -7,7 +7,7 @@ from typing import TYPE_CHECKING
from clan_lib.errors import ClanError
if TYPE_CHECKING:
from .generate import Generator, GeneratorKey
from .generator import Generator, GeneratorKey
class GeneratorNotFoundError(ClanError):

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

@@ -4,8 +4,8 @@ import logging
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 clan_lib.vars.generate import get_generators
from .generate import get_generators
from .generator import Var
log = logging.getLogger(__name__)

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

@@ -8,7 +8,7 @@ from clan_lib.git import commit_files
log = logging.getLogger(__name__)
if TYPE_CHECKING:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
from clan_lib.machines.machines import Machine

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

@@ -146,7 +146,7 @@ class SecretStore(StoreBase):
if not git_hash:
return b""
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
manifest = []
generators = Generator.get_machine_generators(machine, self.flake)
@@ -178,7 +178,7 @@ class SecretStore(StoreBase):
return local_hash != remote_hash.encode()
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators(machine, self.flake)
if "users" in phases:

10
pkgs/clan-cli/clan_cli/vars/secret_modules/sops.py
View File

@@ -23,7 +23,7 @@ from clan_cli.secrets.secrets import (
)
from clan_cli.secrets.sops import load_age_plugins
from clan_cli.vars._types import StoreBase
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
from clan_cli.vars.var import Var
from clan_lib.errors import ClanError
from clan_lib.flake import Flake
@@ -54,7 +54,7 @@ class SecretStore(StoreBase):
def ensure_machine_key(self, machine: str) -> None:
"""Ensure machine has sops keys initialized."""
# no need to generate keys if we don't manage secrets
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators(machine, self.flake)
if not vars_generators:
@@ -135,7 +135,7 @@ class SecretStore(StoreBase):
"""
if generators is None:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators(machine, self.flake)
file_found = False
@@ -212,7 +212,7 @@ class SecretStore(StoreBase):
return [store_folder]
def populate_dir(self, machine: str, output_dir: Path, phases: list[str]) -> None:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
vars_generators = Generator.get_machine_generators(machine, self.flake)
if "users" in phases or "services" in phases:
@@ -347,7 +347,7 @@ class SecretStore(StoreBase):
from clan_cli.secrets.secrets import update_keys
if generators is None:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
generators = Generator.get_machine_generators(machine, self.flake)
file_found = False

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

@@ -3,7 +3,7 @@ from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from clan_cli.vars.generate import Generator
from clan_cli.vars.generator import Generator
from ._types import StoreBase

2
pkgs/clan-cli/clan_cli/vms/run.py
View File

@@ -17,12 +17,12 @@ from clan_lib.dirs import module_root, user_cache_dir, vm_state_dir
from clan_lib.errors import ClanCmdError, ClanError
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_shell
from clan_lib.vars.generate import run_generators
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.facts.generate import generate_facts
from clan_cli.qemu.qga import QgaSession
from clan_cli.qemu.qmp import QEMUMonitorProtocol
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import populate_secret_vars
from .inspect import VmConfig, inspect_vm

2
pkgs/clan-cli/clan_lib/flash/flash.py
View File

@@ -7,7 +7,6 @@ from tempfile import TemporaryDirectory
from typing import Any, Literal
from clan_cli.facts.generate import generate_facts
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import populate_secret_vars
from clan_lib.api import API
@@ -17,6 +16,7 @@ from clan_lib.errors import ClanError
from clan_lib.flake.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_shell
from clan_lib.vars.generate import run_generators
from .automount import pause_automounting
from .list import list_keymaps, list_languages

2
pkgs/clan-cli/clan_lib/machines/install.py
View File

@@ -7,7 +7,6 @@ from typing import Literal
from clan_cli.facts.generate import generate_facts
from clan_cli.machines.hardware import HardwareConfig
from clan_cli.vars.generate import run_generators
from clan_lib.api import API, message_queue
from clan_lib.cmd import Log, RunOpts, run
@@ -15,6 +14,7 @@ from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config, nix_shell
from clan_lib.ssh.create import create_secret_key_nixos_anywhere
from clan_lib.ssh.remote import Remote
from clan_lib.vars.generate import run_generators
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_lib/machines/morph.py
View File

@@ -7,7 +7,6 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from clan_cli.machines.create import CreateOptions, create_machine
from clan_cli.vars.generate import run_generators
from clan_lib.cmd import Log, RunOpts, run
from clan_lib.dirs import specific_machine_dir
@@ -16,6 +15,7 @@ from clan_lib.machines.actions import list_machines
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_build, nix_command
from clan_lib.nix_models.clan import InventoryMachine
from clan_lib.vars.generate import run_generators
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_lib/machines/update.py
View File

@@ -7,7 +7,6 @@ from contextlib import ExitStack
from clan_cli.facts.generate import generate_facts
from clan_cli.facts.upload import upload_secrets
from clan_cli.vars.generate import run_generators
from clan_cli.vars.upload import upload_secret_vars
from clan_lib.api import API
@@ -20,6 +19,7 @@ from clan_lib.nix import nix_command, nix_metadata
from clan_lib.ssh.host import Host
from clan_lib.ssh.localhost import LocalHost
from clan_lib.ssh.remote import Remote
from clan_lib.vars.generate import run_generators
log = logging.getLogger(__name__)

2
pkgs/clan-cli/clan_lib/tests/test_create.py
View File

@@ -14,7 +14,6 @@ from clan_cli.machines.create import create_machine
from clan_cli.secrets.key import generate_key
from clan_cli.secrets.sops import maybe_get_admin_public_keys
from clan_cli.secrets.users import add_user
from clan_cli.vars.generate import get_generators, run_generators
from clan_lib.cmd import RunOpts, run
from clan_lib.dirs import specific_machine_dir
@@ -34,6 +33,7 @@ from clan_lib.persist.util import set_value_by_path
from clan_lib.services.modules import list_service_modules
from clan_lib.ssh.remote import Remote
from clan_lib.templates.disk import hw_main_disk_options, set_machine_disk_schema
from clan_lib.vars.generate import get_generators, run_generators
log = logging.getLogger(__name__)

1
pkgs/clan-cli/clan_lib/vars/__init__.py Normal file
View File

@@ -0,0 +1 @@
# This module contains vars-related functionality

163
pkgs/clan-cli/clan_lib/vars/generate.py Normal file
View File

@@ -0,0 +1,163 @@
import logging
from collections.abc import Callable
from clan_cli.vars.generator import Generator, GeneratorKey
from clan_cli.vars.graph import minimal_closure, requested_closure
from clan_cli.vars.migration import check_can_migrate, migrate_files
from clan_lib.api import API
from clan_lib.errors import ClanError
from clan_lib.machines.machines import Machine
log = logging.getLogger(__name__)
@API.register
def get_generators(
machine: Machine,
full_closure: bool,
generator_name: str | None = None,
include_previous_values: bool = False,
) -> list[Generator]:
"""
Get generators for a machine, with optional closure computation.
Args:
machine: The machine to get generators for.
full_closure: If True, include all dependency generators. If False, only include missing ones.
generator_name: Name of a specific generator to get, or None for all generators.
include_previous_values: If True, populate prompts with their previous values.
Returns:
List of generators based on the specified selection and closure mode.
"""
from clan_cli.vars import graph
vars_generators = Generator.get_machine_generators(machine.name, machine.flake)
generators = {generator.key: generator for generator in vars_generators}
result_closure = []
if generator_name is None: # all generators selected
if full_closure:
result_closure = graph.full_closure(generators)
else:
result_closure = graph.all_missing_closure(generators)
# specific generator selected
elif full_closure:
gen_key = GeneratorKey(machine=machine.name, name=generator_name)
result_closure = requested_closure([gen_key], generators)
else:
gen_key = GeneratorKey(machine=machine.name, name=generator_name)
result_closure = minimal_closure([gen_key], generators)
if include_previous_values:
for generator in result_closure:
for prompt in generator.prompts:
prompt.previous_value = generator.get_previous_value(machine, prompt)
return result_closure
def _ensure_healthy(
machine: "Machine",
generators: list[Generator] | None = None,
) -> None:
"""
Run health checks on the provided generators.
Fails if any of the generators' health checks fail.
"""
if generators is None:
generators = Generator.get_machine_generators(machine.name, machine.flake)
pub_healtcheck_msg = machine.public_vars_store.health_check(
machine.name, generators
)
sec_healtcheck_msg = machine.secret_vars_store.health_check(
machine.name, generators
)
if pub_healtcheck_msg or sec_healtcheck_msg:
msg = f"Health check failed for machine {machine.name}:\n"
if pub_healtcheck_msg:
msg += f"Public vars store: {pub_healtcheck_msg}\n"
if sec_healtcheck_msg:
msg += f"Secret vars store: {sec_healtcheck_msg}"
raise ClanError(msg)
PromptFunc = Callable[[Generator], dict[str, str]]
"""Type for a function that collects prompt values for a generator.
The function receives a Generator and should return a dictionary mapping
prompt names to their values. This allows for custom prompt collection
strategies (e.g., interactive CLI, GUI, or programmatic).
"""
@API.register
def run_generators(
machines: list[Machine],
generators: str | list[str] = "minimal",
prompt_values: dict[str, dict[str, str]] | PromptFunc = lambda g: g.ask_prompts(),
no_sandbox: bool = False,
) -> None:
"""Run the specified generators for machines.
Args:
machines: The machines to run generators for.
generators: Can be:
- str: Single generator name to run (ensuring dependencies are met)
- list[str]: Specific generator names to run exactly as provided.
Dependency generators are not added automatically in this case.
The caller must ensure that all dependencies are included.
- "all": Run all generators (full closure)
- "minimal": Run only missing generators (minimal closure) (default)
prompt_values: A dictionary mapping generator names to their prompt values,
or a function that returns prompt values for a generator.
no_sandbox: Whether to disable sandboxing when executing the generator.
Raises:
ClanError: If the machine or generator is not found, or if there are issues with
executing the generator.
"""
for machine in machines:
if generators == "all":
generator_objects = get_generators(machine, full_closure=True)
elif generators == "minimal":
generator_objects = get_generators(machine, full_closure=False)
elif isinstance(generators, str) and generators not in ["all", "minimal"]:
# Single generator name - compute minimal closure for it
generator_objects = get_generators(
machine, full_closure=False, generator_name=generators
)
elif isinstance(generators, list):
if len(generators) == 0:
return
# Create GeneratorKeys for this specific machine
generator_keys = {
GeneratorKey(machine=machine.name, name=name) for name in generators
}
all_generators = get_generators(machine, full_closure=True)
generator_objects = [g for g in all_generators if g.key in generator_keys]
else:
msg = f"Invalid generators argument: {generators}. Must be 'all', 'minimal', a generator name, or a list of generator names"
raise ValueError(msg)
# If prompt function provided, ask all prompts
# TODO: make this more lazy and ask for every generator on execution
if callable(prompt_values):
prompt_values = {
generator.name: prompt_values(generator)
for generator in generator_objects
}
# execute health check
_ensure_healthy(machine=machine, generators=generator_objects)
# execute generators
for generator in generator_objects:
if check_can_migrate(machine, generator):
migrate_files(machine, generator)
else:
generator.execute(
machine=machine,
prompt_values=prompt_values.get(generator.name, {}),
no_sandbox=no_sandbox,
)

2
pkgs/generate-test-vars/generate_test_vars/cli.py
View File

@@ -11,7 +11,6 @@ from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Any, override
from clan_cli.vars.generate import run_generators
from clan_cli.vars.generator import Generator
from clan_cli.vars.prompt import PromptType
from clan_lib.dirs import find_toplevel
@@ -19,6 +18,7 @@ from clan_lib.errors import ClanError
from clan_lib.flake.flake import Flake
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_config, nix_eval, nix_test_store
from clan_lib.vars.generate import run_generators
log = logging.getLogger(__name__)