API/vars: use string based interfaces to get and set vars to avoid state mutations

This commit is contained in:
Johannes Kirschbauer
2025-01-08 12:49:34 +01:00
committed by hsjobeki
parent 85d03f106b
commit 06869a4d27
6 changed files with 64 additions and 29 deletions

View File

@@ -260,7 +260,11 @@ def _ask_prompts(
prompt_values: dict[str, str] = {}
for prompt in generator.prompts:
var_id = f"{generator.name}/{prompt.name}"
prompt_values[prompt.name] = ask(var_id, prompt.prompt_type)
prompt_values[prompt.name] = ask(
var_id,
prompt.prompt_type,
prompt.description if prompt.description != prompt.name else None,
)
return prompt_values

View File

@@ -2,10 +2,10 @@ import argparse
import logging
import sys
from clan_cli.api import API
from clan_cli.clan_uri import FlakeId
from clan_cli.completions import add_dynamic_completer, complete_machines
from clan_cli.errors import ClanError
from clan_cli.machines.machines import Machine
from .generate import Var
from .list import get_vars
@@ -13,8 +13,9 @@ from .list import get_vars
log = logging.getLogger(__name__)
def get_var(machine: Machine, var_id: str) -> Var:
vars_ = get_vars(machine)
@API.register
def get_var(base_dir: str, machine_name: str, var_id: str) -> Var:
vars_ = get_vars(base_dir=base_dir, machine_name=machine_name)
results = []
for var in vars_:
if var.id == var_id:
@@ -42,8 +43,7 @@ def get_var(machine: Machine, var_id: str) -> Var:
def get_command(machine_name: str, var_id: str, flake: FlakeId) -> None:
machine = Machine(name=machine_name, flake=flake)
var = get_var(machine, var_id)
var = get_var(str(flake.path), machine_name, var_id)
if not var.exists:
msg = f"Var {var.id} has not been generated yet"
raise ClanError(msg)

View File

@@ -25,7 +25,9 @@ def secret_store(machine: Machine) -> StoreBase:
return secret_vars_module.SecretStore(machine=machine)
def get_vars(machine: Machine) -> list[Var]:
@API.register
def get_vars(base_dir: str, machine_name: str) -> list[Var]:
machine = Machine(name=machine_name, flake=FlakeId(base_dir))
pub_store = public_store(machine)
sec_store = secret_store(machine)
all_vars = []
@@ -58,7 +60,7 @@ def _get_previous_value(
@API.register
def get_prompts(base_dir: str, machine_name: str) -> list[Generator]:
def get_generators(base_dir: str, machine_name: str) -> list[Generator]:
machine = Machine(name=machine_name, flake=FlakeId(base_dir))
generators: list[Generator] = machine.vars_generators
for generator in generators:
@@ -96,7 +98,7 @@ def stringify_vars(_vars: list[Var]) -> str:
def stringify_all_vars(machine: Machine) -> str:
return stringify_vars(get_vars(machine))
return stringify_vars(get_vars(str(machine.flake), machine.name))
def list_command(args: argparse.Namespace) -> None:

View File

@@ -37,16 +37,25 @@ class Prompt:
)
def ask(description: str, input_type: PromptType) -> str:
def ask(
ident: str,
input_type: PromptType,
label: str | None,
) -> str:
text = f"Enter the value for {ident}:"
if label:
text = f"{label}"
if MOCK_PROMPT_RESPONSE:
return next(MOCK_PROMPT_RESPONSE)
match input_type:
case PromptType.LINE:
result = input(f"Enter the value for {description}: ")
result = input(f"{text}: ")
case PromptType.MULTILINE:
print(f"Enter the value for {description} (Finish with Ctrl-D): ")
print(f"{text} (Finish with Ctrl-D): ")
result = sys.stdin.read()
case PromptType.HIDDEN:
result = getpass(f"Enter the value for {description} (hidden): ")
result = getpass(f"{text} (hidden): ")
log.info("Input received. Processing...")
return result

View File

@@ -23,7 +23,7 @@ def set_var(
else:
_machine = machine
if isinstance(var, str):
_var = get_var(_machine, var)
_var = get_var(str(flake.path), _machine.name, var)
else:
_var = var
path = _var.set(value)
@@ -36,12 +36,17 @@ def set_var(
def set_via_stdin(machine: str, var_id: str, flake: FlakeId) -> None:
_machine = Machine(name=machine, flake=flake)
var = get_var(_machine, var_id)
var = get_var(str(flake.path), machine, var_id)
if sys.stdin.isatty():
new_value = ask(var.id, PromptType.HIDDEN).encode("utf-8")
new_value = ask(
var.id,
PromptType.HIDDEN,
None,
).encode("utf-8")
else:
new_value = sys.stdin.buffer.read()
_machine = Machine(name=machine, flake=flake)
set_var(_machine, var, new_value, flake)