Merge pull request 'vars: add get_prompts api endpoint' (#2043) from DavHau/clan-core:DavHau-dave into main

pkgs/clan-cli/clan_cli/vars/_types.py

@@ -8,9 +8,26 @@ from pathlib import Path
 from clan_cli.machines.machines import Machine
 
 
+@dataclass
+class Prompt:
+    name: str
+    description: str
+    type: str
+    has_file: bool
+    generator: str
+    previous_value: str | None = None
+
+
+@dataclass
+class Generator:
+    name: str
+    share: bool
+    prompts: list[Prompt]
+
+
 @dataclass
 class Var:
-    store: "StoreBase"
+    _store: "StoreBase"
     generator: str
     name: str
     id: str
@@ -20,11 +37,11 @@ class Var:
 
     @property
     def value(self) -> bytes:
-        if not self.store.exists(self.generator, self.name, self.shared):
+        if not self._store.exists(self.generator, self.name, self.shared):
             msg = f"Var {self.id} has not been generated yet"
             raise ValueError(msg)
         # try decode the value or return <binary blob>
-        return self.store.get(self.generator, self.name, self.shared)
+        return self._store.get(self.generator, self.name, self.shared)
 
     @property
     def printable_value(self) -> str:
@@ -34,16 +51,16 @@ class Var:
             return "<binary blob>"
 
     def set(self, value: bytes) -> None:
-        self.store.set(self.generator, self.name, value, self.shared, self.deployed)
+        self._store.set(self.generator, self.name, value, self.shared, self.deployed)
 
     @property
     def exists(self) -> bool:
-        return self.store.exists(self.generator, self.name, self.shared)
+        return self._store.exists(self.generator, self.name, self.shared)
 
     def __str__(self) -> str:
         if self.secret:
             return f"{self.id}: ********"
-        if self.store.exists(self.generator, self.name, self.shared):
+        if self._store.exists(self.generator, self.name, self.shared):
             return f"{self.id}: {self.printable_value}"
         return f"{self.id}: <not set>"
 
@@ -135,7 +152,7 @@ class StoreBase(ABC):
                     continue
                 all_vars.append(
                     Var(
-                        store=self,
+                        _store=self,
                         generator=gen_name,
                         name=f_name,
                         id=f"{gen_name}/{f_name}",

pkgs/clan-cli/clan_cli/vars/get.py

@@ -8,13 +8,13 @@ from clan_cli.errors import ClanError
 from clan_cli.machines.machines import Machine
 
 from ._types import Var
-from .list import all_vars
+from .list import get_vars
 
 log = logging.getLogger(__name__)
 
 
 def get_var(machine: Machine, var_id: str) -> Var:
-    vars_ = all_vars(machine)
+    vars_ = get_vars(machine)
     results = []
     for var in vars_:
         if var_id in var.id:

pkgs/clan-cli/clan_cli/vars/list.py

@@ -2,21 +2,67 @@ import argparse
 import importlib
 import logging
 
+from clan_cli.api import API
 from clan_cli.completions import add_dynamic_completer, complete_machines
 from clan_cli.machines.machines import Machine
 
-from ._types import Var
+from ._types import Generator, Prompt, StoreBase, Var
 
 log = logging.getLogger(__name__)
 
 
-# TODO get also secret facts
-def all_vars(machine: Machine) -> list[Var]:
+def public_store(machine: Machine) -> StoreBase:
     public_vars_module = importlib.import_module(machine.public_vars_module)
-    public_vars_store = public_vars_module.FactStore(machine=machine)
+    return public_vars_module.FactStore(machine=machine)
+
+
+def secret_store(machine: Machine) -> StoreBase:
     secret_vars_module = importlib.import_module(machine.secret_vars_module)
-    secret_vars_store = secret_vars_module.SecretStore(machine=machine)
-    return public_vars_store.get_all() + secret_vars_store.get_all()
+    return secret_vars_module.SecretStore(machine=machine)
+
+
+def get_vars(machine: Machine) -> list[Var]:
+    pub_store = public_store(machine)
+    sec_store = secret_store(machine)
+    return pub_store.get_all() + sec_store.get_all()
+
+
+def _get_prompt_value(
+    machine: Machine, generator: Generator, prompt: Prompt
+) -> str | None:
+    if not prompt.has_file:
+        return None
+    pub_store = public_store(machine)
+    if pub_store.exists(generator.name, prompt.name, shared=generator.share):
+        return pub_store.get(
+            generator.name, prompt.name, shared=generator.share
+        ).decode()
+    return None
+
+
+@API.register
+def get_prompts(machine: Machine) -> list[Generator]:
+    generators = []
+    for gen_name, generator in machine.vars_generators.items():
+        prompts: list[Prompt] = []
+        gen = Generator(
+            name=gen_name,
+            prompts=prompts,
+            share=generator["share"],
+        )
+        for prompt_name, prompt in generator["prompts"].items():
+            prompt = Prompt(
+                name=prompt_name,
+                description=prompt["description"],
+                type=prompt["type"],
+                has_file=prompt["createFile"],
+                generator=gen_name,
+            )
+            prompt.previous_value = _get_prompt_value(machine, gen, prompt)
+            prompts.append(prompt)
+
+        generators.append(gen)
+    return generators
 
 
 def stringify_vars(_vars: list[Var]) -> str:
@@ -24,7 +70,7 @@ def stringify_vars(_vars: list[Var]) -> str:
 
 
 def stringify_all_vars(machine: Machine) -> str:
-    return stringify_vars(all_vars(machine))
+    return stringify_vars(get_vars(machine))
 
 
 def list_command(args: argparse.Namespace) -> None:

pkgs/clan-cli/tests/test_api_dataclass_compat.py

@@ -119,7 +119,6 @@ def test_all_dataclasses() -> None:
     # - API includes Type Generic wrappers, that are not known in the init file.
     excludes = [
         "api/__init__.py",
-        "vars/_types.py",
     ]
 
     cli_path = Path("clan_cli").resolve()

pkgs/clan-cli/tests/test_vars.py

@@ -409,3 +409,30 @@ def test_prompt_create_file(
     assert sops_store.exists("my_generator", "prompt1")
     assert not sops_store.exists("my_generator", "prompt2")
     assert sops_store.get("my_generator", "prompt1").decode() == "input1"
+
+
+@pytest.mark.impure
+def test_api_get_prompts(
+    monkeypatch: pytest.MonkeyPatch,
+    temporary_home: Path,
+) -> None:
+    from clan_cli.vars.list import get_prompts
+
+    config = nested_dict()
+    my_generator = config["clan"]["core"]["vars"]["generators"]["my_generator"]
+    my_generator["prompts"]["prompt1"]["type"] = "line"
+    my_generator["files"]["prompt1"]["secret"] = False
+    flake = generate_flake(
+        temporary_home,
+        flake_template=CLAN_CORE / "templates" / "minimal",
+        machine_configs={"my_machine": config},
+    )
+    monkeypatch.chdir(flake.path)
+    monkeypatch.setattr("sys.stdin", StringIO("input1"))
+    cli.run(["vars", "generate", "--flake", str(flake.path), "my_machine"])
+    machine = Machine(name="my_machine", flake=FlakeId(str(flake.path)))
+    api_prompts = get_prompts(machine)
+    assert len(api_prompts) == 1
+    assert api_prompts[0].name == "my_generator"
+    assert api_prompts[0].prompts[0].name == "prompt1"
+    assert api_prompts[0].prompts[0].previous_value == "input1"

scripts/pre-commit

@@ -15,7 +15,7 @@ unstash() {
 }
 git stash push --quiet --keep-index --message "pre-commit"
 trap unstash EXIT
-nix fmt
+treefmt
 {
   changed=$(git diff --name-only --exit-code);
   status=$?;