refactor: reduce coupling to Machine class in vars module

- Change Generator class to store machine name as string instead of Machine reference
- Update Generator.generators_from_flake() to only require machine name and flake
- Refactor check_vars() to accept machine name and flake instead of Machine object
- Create Machine instances only when needed for specific operations

This continues the effort to reduce dependencies on the Machine class,
making the codebase more modular and easier to refactor.
This commit is contained in:
DavHau
2025-07-05 15:54:37 +07:00
parent 448e60f866
commit d143359a2d
9 changed files with 50 additions and 61 deletions

View File

@@ -93,9 +93,7 @@ def flash_machine(
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
for generator in Generator.generators_from_flake( for generator in Generator.generators_from_flake(machine.name, machine.flake):
machine.name, machine.flake, machine
):
for file in generator.files: for file in generator.files:
if file.needed_for == "partitioning": if file.needed_for == "partitioning":
msg = f"Partitioning time secrets are not supported with `clan flash write`: clan.core.vars.generators.{generator.name}.files.{file.name}" msg = f"Partitioning time secrets are not supported with `clan flash write`: clan.core.vars.generators.{generator.name}.files.{file.name}"

View File

@@ -118,9 +118,7 @@ def test_add_module_to_inventory(
generator = None generator = None
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(machine.name, machine.flake)
machine.name, machine.flake, machine
)
for gen in generators: for gen in generators:
if gen.name == "borgbackup": if gen.name == "borgbackup":
generator = gen generator = gen

View File

@@ -140,7 +140,7 @@ def test_generate_public_and_secret_vars(
monkeypatch.chdir(flake.path) monkeypatch.chdir(flake.path)
machine = Machine(name="my_machine", flake=Flake(str(flake.path))) machine = Machine(name="my_machine", flake=Flake(str(flake.path)))
assert not check_vars(machine) assert not check_vars(machine.name, machine.flake)
vars_text = stringify_all_vars(machine) vars_text = stringify_all_vars(machine)
assert "my_generator/my_value: <not set>" in vars_text assert "my_generator/my_value: <not set>" in vars_text
assert "my_generator/my_secret: <not set>" in vars_text assert "my_generator/my_secret: <not set>" in vars_text
@@ -158,7 +158,7 @@ def test_generate_public_and_secret_vars(
assert json.loads(value_non_default) == "default_value" assert json.loads(value_non_default) == "default_value"
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"]) cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
assert check_vars(machine) assert check_vars(machine.name, machine.flake)
# get last commit message # get last commit message
commit_message = run( commit_message = run(
["git", "log", "-3", "--pretty=%B"], ["git", "log", "-3", "--pretty=%B"],
@@ -351,10 +351,10 @@ def test_generated_shared_secret_sops(
machine1 = Machine(name="machine1", flake=Flake(str(flake.path))) machine1 = Machine(name="machine1", flake=Flake(str(flake.path)))
machine2 = Machine(name="machine2", flake=Flake(str(flake.path))) machine2 = Machine(name="machine2", flake=Flake(str(flake.path)))
cli.run(["vars", "generate", "--flake", str(flake.path), "machine1"]) cli.run(["vars", "generate", "--flake", str(flake.path), "machine1"])
assert check_vars(machine1) assert check_vars(machine1.name, machine1.flake)
cli.run(["vars", "generate", "--flake", str(flake.path), "machine2"]) cli.run(["vars", "generate", "--flake", str(flake.path), "machine2"])
assert check_vars(machine2) assert check_vars(machine2.name, machine2.flake)
assert check_vars(machine2) assert check_vars(machine2.name, machine2.flake)
m1_sops_store = sops.SecretStore(machine1) m1_sops_store = sops.SecretStore(machine1)
m2_sops_store = sops.SecretStore(machine2) m2_sops_store = sops.SecretStore(machine2)
assert m1_sops_store.exists( assert m1_sops_store.exists(
@@ -404,9 +404,9 @@ def test_generate_secret_var_password_store(
monkeypatch.setenv("PASSWORD_STORE_DIR", str(password_store_dir)) monkeypatch.setenv("PASSWORD_STORE_DIR", str(password_store_dir))
machine = Machine(name="my_machine", flake=Flake(str(flake.path))) machine = Machine(name="my_machine", flake=Flake(str(flake.path)))
assert not check_vars(machine) assert not check_vars(machine.name, machine.flake)
cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"]) cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
assert check_vars(machine) assert check_vars(machine.name, machine.flake)
store = password_store.SecretStore( store = password_store.SecretStore(
Machine(name="my_machine", flake=Flake(str(flake.path))) Machine(name="my_machine", flake=Flake(str(flake.path)))
) )

View File

@@ -4,6 +4,7 @@ import logging
from clan_cli.completions import add_dynamic_completer, complete_machines from clan_cli.completions import add_dynamic_completer, complete_machines
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.flake import Flake
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -26,7 +27,10 @@ class VarStatus:
self.invalid_generators = invalid_generators self.invalid_generators = invalid_generators
def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatus: def vars_status(
machine_name: str, flake: Flake, generator_name: None | str = None
) -> VarStatus:
machine = Machine(name=machine_name, flake=flake)
missing_secret_vars = [] missing_secret_vars = []
missing_public_vars = [] missing_public_vars = []
# signals if a var needs to be updated (eg. needs re-encryption due to new users added) # signals if a var needs to be updated (eg. needs re-encryption due to new users added)
@@ -34,7 +38,7 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
invalid_generators = [] invalid_generators = []
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
generators = Generator.generators_from_flake(machine.name, machine.flake, machine) generators = Generator.generators_from_flake(machine.name, machine.flake)
if generator_name: if generator_name:
for generator in generators: for generator in generators:
if generator_name == generator.name: if generator_name == generator.name:
@@ -47,7 +51,6 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
raise ClanError(err_msg) raise ClanError(err_msg)
for generator in generators: for generator in generators:
generator.machine(machine)
for file in generator.files: for file in generator.files:
file.store( file.store(
machine.secret_vars_store if file.secret else machine.public_vars_store machine.secret_vars_store if file.secret else machine.public_vars_store
@@ -93,8 +96,10 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
) )
def check_vars(machine: Machine, generator_name: None | str = None) -> bool: def check_vars(
status = vars_status(machine, generator_name=generator_name) machine_name: str, flake: Flake, generator_name: None | str = None
) -> bool:
status = vars_status(machine_name, flake, generator_name=generator_name)
return not ( return not (
status.missing_secret_vars status.missing_secret_vars
or status.missing_public_vars or status.missing_public_vars
@@ -104,11 +109,7 @@ def check_vars(machine: Machine, generator_name: None | str = None) -> bool:
def check_command(args: argparse.Namespace) -> None: def check_command(args: argparse.Namespace) -> None:
machine = Machine( ok = check_vars(args.machine, args.flake, generator_name=args.generator)
name=args.machine,
flake=args.flake,
)
ok = check_vars(machine, generator_name=args.generator)
if not ok: if not ok:
raise SystemExit(1) raise SystemExit(1)

View File

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

View File

@@ -49,20 +49,17 @@ class Generator:
migrate_fact: str | None = None migrate_fact: str | None = None
# TODO: remove this machine: str | None = None
_machine: "Machine | None" = None _flake: "Flake | None" = None
def machine(self, machine: "Machine") -> None:
self._machine = machine
@cached_property @cached_property
def exists(self) -> bool: def exists(self) -> bool:
assert self._machine is not None assert self.machine is not None and self._flake is not None
return check_vars(self._machine, generator_name=self.name) return check_vars(self.machine, self._flake, generator_name=self.name)
@classmethod @classmethod
def generators_from_flake( def generators_from_flake(
cls: type["Generator"], machine_name: str, flake: "Flake", machine: "Machine" cls: type["Generator"], machine_name: str, flake: "Flake"
) -> list["Generator"]: ) -> list["Generator"]:
config = nix_config() config = nix_config()
system = config["system"] system = config["system"]
@@ -112,19 +109,21 @@ class Generator:
dependencies=gen_data["dependencies"], dependencies=gen_data["dependencies"],
migrate_fact=gen_data.get("migrateFact"), migrate_fact=gen_data.get("migrateFact"),
prompts=prompts, prompts=prompts,
machine=machine_name,
_flake=flake,
) )
# Set the machine immediately
generator.machine(machine)
generators.append(generator) generators.append(generator)
return generators return generators
def final_script(self) -> Path: def final_script(self) -> Path:
assert self._machine is not None assert self.machine is not None and self._flake is not None
from clan_lib.machines.machines import Machine
from clan_lib.nix import nix_test_store from clan_lib.nix import nix_test_store
machine = Machine(name=self.machine, flake=self._flake)
output = Path( output = Path(
self._machine.select( machine.select(
f'config.clan.core.vars.generators."{self.name}".finalScript' f'config.clan.core.vars.generators."{self.name}".finalScript'
) )
) )
@@ -133,8 +132,11 @@ class Generator:
return output return output
def validation(self) -> str | None: def validation(self) -> str | None:
assert self._machine is not None assert self.machine is not None and self._flake is not None
return self._machine.select( 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' f'config.clan.core.vars.generators."{self.name}".validationHash'
) )
@@ -187,9 +189,7 @@ def decrypt_dependencies(
decrypted_dependencies: dict[str, Any] = {} decrypted_dependencies: dict[str, Any] = {}
for generator_name in set(generator.dependencies): for generator_name in set(generator.dependencies):
decrypted_dependencies[generator_name] = {} decrypted_dependencies[generator_name] = {}
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(machine.name, machine.flake)
machine.name, machine.flake, machine
)
for dep_generator in generators: for dep_generator in generators:
if generator_name == dep_generator.name: if generator_name == dep_generator.name:
break break
@@ -398,9 +398,7 @@ def get_closure(
) -> list[Generator]: ) -> list[Generator]:
from . import graph from . import graph
vars_generators = Generator.generators_from_flake( vars_generators = Generator.generators_from_flake(machine.name, machine.flake)
machine.name, machine.flake, machine
)
generators: dict[str, Generator] = { generators: dict[str, Generator] = {
generator.name: generator for generator in vars_generators generator.name: generator for generator in vars_generators
} }
@@ -477,7 +475,7 @@ def generate_vars_for_machine(
generators_set = set(generators) generators_set = set(generators)
generators_ = [ generators_ = [
g g
for g in Generator.generators_from_flake(machine_name, machine.flake, machine) for g in Generator.generators_from_flake(machine_name, machine.flake)
if g.name in generators_set if g.name in generators_set
] ]
@@ -498,9 +496,7 @@ def generate_vars_for_machine_interactive(
) -> bool: ) -> bool:
_generator = None _generator = None
if generator_name: if generator_name:
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(machine.name, machine.flake)
machine.name, machine.flake, machine
)
for generator in generators: for generator in generators:
if generator.name == generator_name: if generator.name == generator_name:
_generator = generator _generator = generator

View File

@@ -21,9 +21,7 @@ def get_vars(base_dir: str, machine_name: str) -> list[Var]:
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
all_vars = [] all_vars = []
for generator in Generator.generators_from_flake( for generator in Generator.generators_from_flake(machine_name, machine.flake):
machine_name, machine.flake, machine
):
for var in generator.files: for var in generator.files:
if var.secret: if var.secret:
var.store(sec_store) var.store(sec_store)
@@ -57,7 +55,7 @@ def get_generators(base_dir: str, machine_name: str) -> list[Generator]:
machine = Machine(name=machine_name, flake=Flake(base_dir)) machine = Machine(name=machine_name, flake=Flake(base_dir))
generators: list[Generator] = Generator.generators_from_flake( generators: list[Generator] = Generator.generators_from_flake(
machine_name, machine.flake, machine machine_name, machine.flake
) )
for generator in generators: for generator in generators:
for prompt in generator.prompts: for prompt in generator.prompts:
@@ -76,9 +74,7 @@ def set_prompts(
machine = Machine(name=machine_name, flake=Flake(base_dir)) machine = Machine(name=machine_name, flake=Flake(base_dir))
for update in updates: for update in updates:
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(machine_name, machine.flake)
machine_name, machine.flake, machine
)
for generator in generators: for generator in generators:
if generator.name == update.generator: if generator.name == update.generator:
break break

View File

@@ -144,7 +144,7 @@ class SecretStore(StoreBase):
manifest = [] manifest = []
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
for generator in generators: for generator in generators:
for file in generator.files: for file in generator.files:
@@ -173,7 +173,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake( vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
if "users" in phases: if "users" in phases:
with tarfile.open( with tarfile.open(

View File

@@ -55,7 +55,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake( vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
if not vars_generators: if not vars_generators:
return return
@@ -118,7 +118,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
else: else:
generators = [generator] generators = [generator]
@@ -195,7 +195,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
vars_generators = Generator.generators_from_flake( vars_generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
if "users" in phases or "services" in phases: if "users" in phases or "services" in phases:
key_name = f"{self.machine.name}-age.key" key_name = f"{self.machine.name}-age.key"
@@ -310,7 +310,7 @@ class SecretStore(StoreBase):
from clan_cli.vars.generate import Generator from clan_cli.vars.generate import Generator
generators = Generator.generators_from_flake( generators = Generator.generators_from_flake(
self.machine.name, self.machine.flake, self.machine self.machine.name, self.machine.flake
) )
else: else:
generators = [generator] generators = [generator]