Merge pull request 'vars: set vars via cli; improve getting vars via cli;' (#2039) from DavHau/clan-core:DavHau-dave into main

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

@@ -7,6 +7,7 @@ from .check import register_check_parser
 from .generate import register_generate_parser
 from .get import register_get_parser
 from .list import register_list_parser
+from .set import register_set_parser
 from .upload import register_upload_parser
 
 
@@ -85,6 +86,25 @@ For more detailed information, visit: {help_hyperlink("secrets", "https://docs.c
     )
     register_get_parser(get_parser)
 
+    set_parser = subparser.add_parser(
+        "set",
+        help="set a specific var",
+        epilog=(
+            f"""
+This subcommand allows setting a specific var for a specific machine.
+
+Examples:
+
+    $ clan vars set my-server zerotier/vpn-ip
+    Will set the var for the specified machine.
+
+For more detailed information, visit: {help_hyperlink("secrets", "https://docs.clan.lol/getting-started/secrets")}
+        """
+        ),
+        formatter_class=argparse.RawTextHelpFormatter,
+    )
+    register_set_parser(set_parser)
+
     parser_generate = subparser.add_parser(
         "generate",
         help="(re-)generate vars for specific or all machines",

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

@@ -1,5 +1,6 @@
 # !/usr/bin/env python3
 import json
+import shutil
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from pathlib import Path
@@ -32,6 +33,9 @@ class Var:
         except UnicodeDecodeError:
             return "<binary blob>"
 
+    def set(self, value: bytes) -> None:
+        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)
@@ -109,8 +113,7 @@ class StoreBase(ABC):
         directory = self.directory(generator_name, var_name, shared)
         # delete directory
         if directory.exists():
-            for f in directory.glob("*"):
+            shutil.rmtree(directory)
-                f.unlink()
         # re-create directory
         directory.mkdir(parents=True, exist_ok=True)
         new_file = self._set(generator_name, var_name, value, shared, deployed)

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

@@ -3,7 +3,6 @@ import importlib
 import logging
 import os
 import sys
-from getpass import getpass
 from graphlib import TopologicalSorter
 from pathlib import Path
 from tempfile import TemporaryDirectory
@@ -22,6 +21,7 @@ from clan_cli.machines.machines import Machine
 from clan_cli.nix import nix_shell
 
 from .check import check_vars
+from .prompt import prompt
 from .public_modules import FactStoreBase
 from .secret_modules import SecretStoreBase
 
@@ -133,11 +133,11 @@ def execute_generator(
         if machine.vars_generators[generator_name]["prompts"]:
             tmpdir_prompts.mkdir()
             env["prompts"] = str(tmpdir_prompts)
-            for prompt_name, prompt in machine.vars_generators[generator_name][
+            for prompt_name, prompt_ in machine.vars_generators[generator_name][
                 "prompts"
             ].items():
                 prompt_file = tmpdir_prompts / prompt_name
-                value = prompt_func(prompt["description"], prompt["type"])
+                value = prompt(prompt_["description"], prompt_["type"])
                 prompt_file.write_text(value)
 
         if sys.platform == "linux":
@@ -184,21 +184,6 @@ def execute_generator(
     return True
 
 
-def prompt_func(description: str, input_type: str) -> str:
-    if input_type == "line":
-        result = input(f"Enter the value for {description}: ")
-    elif input_type == "multiline":
-        print(f"Enter the value for {description} (Finish with Ctrl-D): ")
-        result = sys.stdin.read()
-    elif input_type == "hidden":
-        result = getpass(f"Enter the value for {description} (hidden): ")
-    else:
-        msg = f"Unknown input type: {input_type} for prompt {description}"
-        raise ClanError(msg)
-    log.info("Input received. Processing...")
-    return result
-
-
 def _get_subgraph(graph: dict[str, set], vertex: str) -> dict[str, set]:
     visited = set()
     queue = [vertex]

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

@@ -13,14 +13,15 @@ from .list import all_vars
 log = logging.getLogger(__name__)
 
 
-def get_var(machine: Machine, var_id: str) -> Var | None:
+def get_var(machine: Machine, var_id: str) -> Var:
     vars_ = all_vars(machine)
     results = []
     for var in vars_:
         if var_id in var.id:
             results.append(var)
     if len(results) == 0:
-        return None
+        msg = f"No var found for search string: {var_id}"
+        raise ClanError(msg)
     if len(results) > 1:
         error = (
             f"Found multiple vars for {var_id}:\n  - "
@@ -29,28 +30,33 @@ def get_var(machine: Machine, var_id: str) -> Var | None:
         )
         raise ClanError(error)
     # we have exactly one result at this point
-    result = results[0]
+    var = results[0]
-    if var_id == result.id:
+    if var_id == var.id:
-        return result
+        return var
-    msg = f"Did you mean: {result.id}"
+    msg = f"Did you mean: {var.id}"
     raise ClanError(msg)
 
 
-def get_command(
+def get_command(machine: str, var_id: str, flake: FlakeId) -> None:
-    machine: str, var_id: str, flake: FlakeId, quiet: bool, **kwargs: dict
-) -> None:
     _machine = Machine(name=machine, flake=flake)
     var = get_var(_machine, var_id)
-    if var is None:
-        msg = f"No var found for search string: {var_id}"
-        raise ClanError(msg)
     if not var.exists:
         msg = f"Var {var.id} has not been generated yet"
         raise ClanError(msg)
-    if quiet:
+    if sys.stdout.isatty():
         sys.stdout.buffer.write(var.value)
     else:
-        print(f"{var.id}: {var.printable_value}")
+        print(var.printable_value)
+
+
+def _get_command(
+    args: argparse.Namespace,
+) -> None:
+    get_command(
+        machine=args.machine,
+        var_id=args.var_id,
+        flake=args.flake,
+    )
 
 
 def register_get_parser(parser: argparse.ArgumentParser) -> None:
@@ -65,10 +71,4 @@ def register_get_parser(parser: argparse.ArgumentParser) -> None:
         help="The var id to get the value for. Example: ssh-keys/pubkey",
     )
 
-    parser.add_argument(
+    parser.set_defaults(func=_get_command)
-        "--quiet",
-        "-q",
-        help="Only print the value of the var",
-        action="store_true",
-    )
-    parser.set_defaults(func=lambda args: get_command(**vars(args)))

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

@@ -0,0 +1,22 @@
+import logging
+import sys
+from getpass import getpass
+
+from clan_cli.errors import ClanError
+
+log = logging.getLogger(__name__)
+
+
+def prompt(description: str, input_type: str) -> str:
+    if input_type == "line":
+        result = input(f"Enter the value for {description}: ")
+    elif input_type == "multiline":
+        print(f"Enter the value for {description} (Finish with Ctrl-D): ")
+        result = sys.stdin.read()
+    elif input_type == "hidden":
+        result = getpass(f"Enter the value for {description} (hidden): ")
+    else:
+        msg = f"Unknown input type: {input_type} for prompt {description}"
+        raise ClanError(msg)
+    log.info("Input received. Processing...")
+    return result

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

@@ -0,0 +1,41 @@
+import argparse
+import logging
+import sys
+
+from clan_cli.clan_uri import FlakeId
+from clan_cli.completions import add_dynamic_completer, complete_machines
+from clan_cli.machines.machines import Machine
+from clan_cli.vars.get import get_var
+
+from .prompt import prompt
+
+log = logging.getLogger(__name__)
+
+
+def get_command(machine: str, var_id: str, flake: FlakeId) -> None:
+    _machine = Machine(name=machine, flake=flake)
+    var = get_var(_machine, var_id)
+    if sys.stdin.isatty():
+        new_value = prompt(var.id, "hidden").encode("utf-8")
+    else:
+        new_value = sys.stdin.buffer.read()
+    var.set(new_value)
+
+
+def _get_command(args: argparse.Namespace) -> None:
+    get_command(args.machine, args.var_id, args.flake)
+
+
+def register_set_parser(parser: argparse.ArgumentParser) -> None:
+    machines_arg = parser.add_argument(
+        "machine",
+        help="The machine to set a var for",
+    )
+    add_dynamic_completer(machines_arg, complete_machines)
+
+    parser.add_argument(
+        "var_id",
+        help="The var id for which to set the value. Example: ssh-keys/pubkey",
+    )
+
+    parser.set_defaults(func=_get_command)