refactor: remove Machine.vars_generators() method
Replace all calls to machine.vars_generators() with direct calls to Generator.generators_from_flake() to make the dependency more explicit and remove unnecessary indirection. This reduces coupling to the Machine class, making the codebase more modular and easier to refactor in the future.
This commit is contained in:
@@ -91,7 +91,11 @@ def flash_machine(
|
|||||||
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
|
"users": {"root": {"openssh": {"authorizedKeys": {"keys": root_keys}}}}
|
||||||
}
|
}
|
||||||
|
|
||||||
for generator in machine.vars_generators():
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
for generator in Generator.generators_from_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}"
|
||||||
|
|||||||
@@ -114,9 +114,14 @@ def test_add_module_to_inventory(
|
|||||||
name="machine1", flake=Flake(str(test_flake_with_core.path))
|
name="machine1", flake=Flake(str(test_flake_with_core.path))
|
||||||
)
|
)
|
||||||
|
|
||||||
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
generator = None
|
generator = None
|
||||||
|
|
||||||
for gen in machine.vars_generators():
|
generators = Generator.generators_from_flake(
|
||||||
|
machine.name, machine.flake, machine
|
||||||
|
)
|
||||||
|
for gen in generators:
|
||||||
if gen.name == "borgbackup":
|
if gen.name == "borgbackup":
|
||||||
generator = gen
|
generator = gen
|
||||||
break
|
break
|
||||||
|
|||||||
@@ -32,7 +32,9 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
|
|||||||
# 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)
|
||||||
unfixed_secret_vars = []
|
unfixed_secret_vars = []
|
||||||
invalid_generators = []
|
invalid_generators = []
|
||||||
generators = machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
generators = Generator.generators_from_flake(machine.name, machine.flake, machine)
|
||||||
if generator_name:
|
if generator_name:
|
||||||
for generator in generators:
|
for generator in generators:
|
||||||
if generator_name == generator.name:
|
if generator_name == generator.name:
|
||||||
|
|||||||
@@ -9,7 +9,9 @@ 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:
|
||||||
generators = machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
generators = Generator.generators_from_flake(machine.name, machine.flake, machine)
|
||||||
if generator_name:
|
if generator_name:
|
||||||
for generator in generators:
|
for generator in generators:
|
||||||
if generator_name == generator.name:
|
if generator_name == generator.name:
|
||||||
|
|||||||
@@ -187,7 +187,10 @@ 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] = {}
|
||||||
for dep_generator in machine.vars_generators():
|
generators = Generator.generators_from_flake(
|
||||||
|
machine.name, machine.flake, machine
|
||||||
|
)
|
||||||
|
for dep_generator in generators:
|
||||||
if generator_name == dep_generator.name:
|
if generator_name == dep_generator.name:
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
@@ -395,7 +398,9 @@ def get_closure(
|
|||||||
) -> list[Generator]:
|
) -> list[Generator]:
|
||||||
from . import graph
|
from . import graph
|
||||||
|
|
||||||
vars_generators = machine.vars_generators()
|
vars_generators = Generator.generators_from_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
|
||||||
}
|
}
|
||||||
@@ -470,7 +475,11 @@ def generate_vars_for_machine(
|
|||||||
|
|
||||||
machine = Machine(name=machine_name, flake=Flake(str(base_dir)))
|
machine = Machine(name=machine_name, flake=Flake(str(base_dir)))
|
||||||
generators_set = set(generators)
|
generators_set = set(generators)
|
||||||
generators_ = [g for g in machine.vars_generators() if g.name in generators_set]
|
generators_ = [
|
||||||
|
g
|
||||||
|
for g in Generator.generators_from_flake(machine_name, machine.flake, machine)
|
||||||
|
if g.name in generators_set
|
||||||
|
]
|
||||||
|
|
||||||
return _generate_vars_for_machine(
|
return _generate_vars_for_machine(
|
||||||
machine=machine,
|
machine=machine,
|
||||||
@@ -489,7 +498,10 @@ def generate_vars_for_machine_interactive(
|
|||||||
) -> bool:
|
) -> bool:
|
||||||
_generator = None
|
_generator = None
|
||||||
if generator_name:
|
if generator_name:
|
||||||
for generator in machine.vars_generators():
|
generators = Generator.generators_from_flake(
|
||||||
|
machine.name, machine.flake, machine
|
||||||
|
)
|
||||||
|
for generator in generators:
|
||||||
if generator.name == generator_name:
|
if generator.name == generator_name:
|
||||||
_generator = generator
|
_generator = generator
|
||||||
break
|
break
|
||||||
|
|||||||
@@ -18,8 +18,12 @@ def get_vars(base_dir: str, machine_name: str) -> list[Var]:
|
|||||||
machine = Machine(name=machine_name, flake=Flake(base_dir))
|
machine = Machine(name=machine_name, flake=Flake(base_dir))
|
||||||
pub_store = machine.public_vars_store
|
pub_store = machine.public_vars_store
|
||||||
sec_store = machine.secret_vars_store
|
sec_store = machine.secret_vars_store
|
||||||
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
all_vars = []
|
all_vars = []
|
||||||
for generator in machine.vars_generators():
|
for generator in Generator.generators_from_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)
|
||||||
@@ -49,8 +53,12 @@ def _get_previous_value(
|
|||||||
|
|
||||||
@API.register
|
@API.register
|
||||||
def get_generators(base_dir: str, machine_name: str) -> list[Generator]:
|
def get_generators(base_dir: str, machine_name: str) -> list[Generator]:
|
||||||
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
machine = Machine(name=machine_name, flake=Flake(base_dir))
|
machine = Machine(name=machine_name, flake=Flake(base_dir))
|
||||||
generators: list[Generator] = machine.vars_generators()
|
generators: list[Generator] = Generator.generators_from_flake(
|
||||||
|
machine_name, machine.flake, machine
|
||||||
|
)
|
||||||
for generator in generators:
|
for generator in generators:
|
||||||
for prompt in generator.prompts:
|
for prompt in generator.prompts:
|
||||||
prompt.previous_value = _get_previous_value(machine, generator, prompt)
|
prompt.previous_value = _get_previous_value(machine, generator, prompt)
|
||||||
@@ -64,9 +72,14 @@ def get_generators(base_dir: str, machine_name: str) -> list[Generator]:
|
|||||||
def set_prompts(
|
def set_prompts(
|
||||||
base_dir: str, machine_name: str, updates: list[GeneratorUpdate]
|
base_dir: str, machine_name: str, updates: list[GeneratorUpdate]
|
||||||
) -> None:
|
) -> None:
|
||||||
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
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:
|
||||||
for generator in machine.vars_generators():
|
generators = Generator.generators_from_flake(
|
||||||
|
machine_name, machine.flake, machine
|
||||||
|
)
|
||||||
|
for generator in generators:
|
||||||
if generator.name == update.generator:
|
if generator.name == update.generator:
|
||||||
break
|
break
|
||||||
else:
|
else:
|
||||||
|
|||||||
@@ -140,8 +140,13 @@ class SecretStore(StoreBase):
|
|||||||
# we sort the hashes to make sure that the order is always the same
|
# we sort the hashes to make sure that the order is always the same
|
||||||
hashes.sort()
|
hashes.sort()
|
||||||
|
|
||||||
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
manifest = []
|
manifest = []
|
||||||
for generator in self.machine.vars_generators():
|
generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
|
for generator in generators:
|
||||||
for file in generator.files:
|
for file in generator.files:
|
||||||
manifest.append(f"{generator.name}/{file.name}".encode())
|
manifest.append(f"{generator.name}/{file.name}".encode())
|
||||||
manifest += hashes
|
manifest += hashes
|
||||||
@@ -165,7 +170,11 @@ class SecretStore(StoreBase):
|
|||||||
return local_hash.decode() != remote_hash
|
return local_hash.decode() != remote_hash
|
||||||
|
|
||||||
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
|
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
|
||||||
vars_generators = self.machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
vars_generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
if "users" in phases:
|
if "users" in phases:
|
||||||
with tarfile.open(
|
with tarfile.open(
|
||||||
output_dir / "secrets_for_users.tar.gz", "w:gz"
|
output_dir / "secrets_for_users.tar.gz", "w:gz"
|
||||||
|
|||||||
@@ -52,7 +52,11 @@ class SecretStore(StoreBase):
|
|||||||
self.machine = machine
|
self.machine = machine
|
||||||
|
|
||||||
# no need to generate keys if we don't manage secrets
|
# no need to generate keys if we don't manage secrets
|
||||||
vars_generators = self.machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
vars_generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
if not vars_generators:
|
if not vars_generators:
|
||||||
return
|
return
|
||||||
has_secrets = False
|
has_secrets = False
|
||||||
@@ -111,7 +115,11 @@ class SecretStore(StoreBase):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
if generator is None:
|
if generator is None:
|
||||||
generators = self.machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
generators = [generator]
|
generators = [generator]
|
||||||
file_found = False
|
file_found = False
|
||||||
@@ -184,7 +192,11 @@ class SecretStore(StoreBase):
|
|||||||
return [store_folder]
|
return [store_folder]
|
||||||
|
|
||||||
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
|
def populate_dir(self, output_dir: Path, phases: list[str]) -> None:
|
||||||
vars_generators = self.machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
vars_generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
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"
|
||||||
if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):
|
if not has_secret(sops_secrets_folder(self.machine.flake_dir) / key_name):
|
||||||
@@ -295,7 +307,11 @@ class SecretStore(StoreBase):
|
|||||||
from clan_cli.secrets.secrets import update_keys
|
from clan_cli.secrets.secrets import update_keys
|
||||||
|
|
||||||
if generator is None:
|
if generator is None:
|
||||||
generators = self.machine.vars_generators()
|
from clan_cli.vars.generate import Generator
|
||||||
|
|
||||||
|
generators = Generator.generators_from_flake(
|
||||||
|
self.machine.name, self.machine.flake, self.machine
|
||||||
|
)
|
||||||
else:
|
else:
|
||||||
generators = [generator]
|
generators = [generator]
|
||||||
file_found = False
|
file_found = False
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ from clan_lib.ssh.remote import Remote
|
|||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
from clan_cli.vars.generate import Generator
|
pass
|
||||||
|
|
||||||
|
|
||||||
@dataclass(frozen=True)
|
@dataclass(frozen=True)
|
||||||
@@ -118,11 +118,6 @@ class Machine:
|
|||||||
return services
|
return services
|
||||||
return {}
|
return {}
|
||||||
|
|
||||||
def vars_generators(self) -> list["Generator"]:
|
|
||||||
from clan_cli.vars.generate import Generator
|
|
||||||
|
|
||||||
return Generator.generators_from_flake(self.name, self.flake, self)
|
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def secrets_upload_directory(self) -> str:
|
def secrets_upload_directory(self) -> str:
|
||||||
return self.select("config.clan.core.facts.secretUploadDirectory")
|
return self.select("config.clan.core.facts.secretUploadDirectory")
|
||||||
|
|||||||
Reference in New Issue
Block a user