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

use machine.{secrets,public}_{vars,fact}_store everywhere

Browse Source
This commit is contained in:
Jörg Thalheim
2025-04-22 16:53:31 +02:00
parent cbde58e1d8
commit 3ac1907201
12 changed files with 39 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,9 @@
import argparse
import importlib
import logging
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.errors import ClanError
from clan_cli.machines.machines import Machine
from clan_cli.vars._types import StoreBase
log = logging.getLogger(__name__)
@@ -30,11 +28,6 @@ class VarStatus:
def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatus:
secret_vars_module = importlib.import_module(machine.secret_vars_module)
secret_vars_store: StoreBase = secret_vars_module.SecretStore(machine=machine)
public_vars_module = importlib.import_module(machine.public_vars_module)
public_vars_store: StoreBase = public_vars_module.FactStore(machine=machine)
missing_secret_vars = []
missing_public_vars = []
# signals if a var needs to be updated (eg. needs re-encryption due to new users added)
@@ -55,17 +48,19 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
for generator in generators:
generator.machine(machine)
for file in generator.files:
file.store(secret_vars_store if file.secret else public_vars_store)
file.store(
machine.secret_vars_store if file.secret else machine.public_vars_store
)
file.generator(generator)
if file.secret:
if not secret_vars_store.exists(generator, file.name):
if not machine.secret_vars_store.exists(generator, file.name):
machine.info(
f"Secret var '{file.name}' for service '{generator.name}' in machine {machine.name} is missing."
)
missing_secret_vars.append(file)
else:
msg = secret_vars_store.health_check(
msg = machine.secret_vars_store.health_check(
generator=generator,
file_name=file.name,
)
@@ -75,15 +70,15 @@ def vars_status(machine: Machine, generator_name: None | str = None) -> VarStatu
)
unfixed_secret_vars.append(file)
elif not public_vars_store.exists(generator, file.name):
elif not machine.public_vars_store.exists(generator, file.name):
machine.info(
f"Public var '{file.name}' for service '{generator.name}' in machine {machine.name} is missing."
)
missing_public_vars.append(file)
# check if invalidation hash is up to date
if not (
secret_vars_store.hash_is_valid(generator)
and public_vars_store.hash_is_valid(generator)
machine.secret_vars_store.hash_is_valid(generator)
and machine.public_vars_store.hash_is_valid(generator)
):
invalid_generators.append(generator.name)
machine.info(

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

@@ -1,21 +1,14 @@
import argparse
import importlib
import logging
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.errors import ClanError
from clan_cli.machines.machines import Machine
from clan_cli.vars._types import StoreBase
log = logging.getLogger(__name__)
def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
secret_vars_module = importlib.import_module(machine.secret_vars_module)
secret_vars_store: StoreBase = secret_vars_module.SecretStore(machine=machine)
public_vars_module = importlib.import_module(machine.public_vars_module)
public_vars_store: StoreBase = public_vars_module.FactStore(machine=machine)
generators = machine.vars_generators
if generator_name:
for generator in generators:
@@ -29,8 +22,8 @@ def fix_vars(machine: Machine, generator_name: None | str = None) -> None:
raise ClanError(err_msg)
for generator in generators:
public_vars_store.fix(generator=generator)
secret_vars_store.fix(generator=generator)
machine.public_vars_store.fix(generator=generator)
machine.secret_vars_store.fix(generator=generator)
def fix_command(args: argparse.Namespace) -> None:

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

@@ -1,5 +1,4 @@
import argparse
import importlib
import logging
from clan_cli.api import API
@@ -7,7 +6,6 @@ from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.errors import ClanError
from clan_cli.flake import Flake
from clan_cli.machines.machines import Machine
from clan_cli.vars._types import StoreBase
from ._types import GeneratorUpdate
from .generate import Generator, Prompt, Var, execute_generator
@@ -15,21 +13,11 @@ from .generate import Generator, Prompt, Var, execute_generator
log = logging.getLogger(__name__)
def public_store(machine: Machine) -> StoreBase:
public_vars_module = importlib.import_module(machine.public_vars_module)
return public_vars_module.FactStore(machine=machine)
def secret_store(machine: Machine) -> StoreBase:
secret_vars_module = importlib.import_module(machine.secret_vars_module)
return secret_vars_module.SecretStore(machine=machine)
@API.register
def get_vars(base_dir: str, machine_name: str) -> list[Var]:
machine = Machine(name=machine_name, flake=Flake(base_dir))
pub_store = public_store(machine)
sec_store = secret_store(machine)
pub_store = machine.public_vars_store
sec_store = machine.secret_vars_store
all_vars = []
for generator in machine.vars_generators:
for var in generator.files:
@@ -50,10 +38,10 @@ def _get_previous_value(
if not prompt.persist:
return None
pub_store = public_store(machine)
pub_store = machine.public_vars_store
if pub_store.exists(generator, prompt.name):
return pub_store.get(generator, prompt.name).decode()
sec_store = secret_store(machine)
sec_store = machine.secret_vars_store
if sec_store.exists(generator, prompt.name):
return sec_store.get(generator, prompt.name).decode()
return None
@@ -87,8 +75,8 @@ def set_prompts(
execute_generator(
machine,
generator,
secret_vars_store=secret_store(machine),
public_vars_store=public_store(machine),
secret_vars_store=machine.secret_vars_store,
public_vars_store=machine.public_vars_store,
prompt_values=update.prompt_values,
)

9
pkgs/clan-cli/clan_cli/vars/upload.py
View File

@@ -1,5 +1,4 @@
import argparse
import importlib
import logging
from pathlib import Path
@@ -10,12 +9,12 @@ log = logging.getLogger(__name__)
def upload_secret_vars(machine: Machine, directory: Path | None = None) -> None:
secret_store_module = importlib.import_module(machine.secret_vars_module)
secret_store = secret_store_module.SecretStore(machine=machine)
if directory:
secret_store.populate_dir(directory, phases=["activation", "users", "services"])
machine.secret_vars_store.populate_dir(
directory, phases=["activation", "users", "services"]
)
else:
secret_store.upload(phases=["activation", "users", "services"])
machine.secret_vars_store.upload(phases=["activation", "users", "services"])
def upload_command(args: argparse.Namespace) -> None: