vars: add 'get' command to cli

This commit is contained in:
DavHau
2024-09-03 19:17:42 +02:00
parent 004a1ba45a
commit e3280e2b1d
4 changed files with 126 additions and 9 deletions

View File

@@ -5,6 +5,7 @@ from clan_cli.hyperlink import help_hyperlink
from .check import register_check_parser from .check import register_check_parser
from .generate import register_generate_parser from .generate import register_generate_parser
from .get import register_get_parser
from .list import register_list_parser from .list import register_list_parser
from .upload import register_upload_parser from .upload import register_upload_parser
@@ -65,6 +66,25 @@ For more detailed information, visit: {help_hyperlink("secrets", "https://docs.c
) )
register_list_parser(list_parser) register_list_parser(list_parser)
get_parser = subparser.add_parser(
"get",
help="get a specific var",
epilog=(
f"""
This subcommand allows getting a specific var for a specific machine.
Examples:
$ clan vars get my-server zerotier/vpn-ip
Will get 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_get_parser(get_parser)
parser_generate = subparser.add_parser( parser_generate = subparser.add_parser(
"generate", "generate",
help="(re-)generate vars for specific or all machines", help="(re-)generate vars for specific or all machines",

View File

@@ -12,14 +12,36 @@ class Var:
store: "StoreBase" store: "StoreBase"
generator: str generator: str
name: str name: str
id: str
secret: bool secret: bool
shared: bool shared: bool
deployed: bool deployed: bool
@property
def value(self) -> bytes:
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)
@property
def printable_value(self) -> str:
try:
return self.value.decode()
except UnicodeDecodeError:
return "<binary blob>"
@property
def exists(self) -> bool:
return self.store.exists(self.generator, self.name, self.shared)
def __str__(self) -> str: def __str__(self) -> str:
if self.secret: if self.secret:
return f"{self.generator}/{self.name}: ********" return f"{self.id}: ********"
return f"{self.generator}/{self.name}: {self.store.get(self.generator, self.name, self.shared).decode()}" if self.store.exists(self.generator, self.name, self.shared):
return f"{self.id}: {self.printable_value}"
return f"{self.id}: <not set>"
class StoreBase(ABC): class StoreBase(ABC):
@@ -113,6 +135,7 @@ class StoreBase(ABC):
store=self, store=self,
generator=gen_name, generator=gen_name,
name=f_name, name=f_name,
id=f"{gen_name}/{f_name}",
secret=file["secret"], secret=file["secret"],
shared=generator["share"], shared=generator["share"],
deployed=file["deploy"], deployed=file["deploy"],

View File

@@ -0,0 +1,74 @@
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.errors import ClanError
from clan_cli.machines.machines import Machine
from ._types import Var
from .list import all_vars
log = logging.getLogger(__name__)
def get_var(machine: Machine, var_id: str) -> Var | None:
vars_ = all_vars(machine)
results = []
for var in vars_:
if var_id in var.id:
results.append(var)
if len(results) == 0:
return None
if len(results) > 1:
error = (
f"Found multiple vars for {var_id}:\n - "
+ "\n - ".join([str(var) for var in results])
+ "\nBe more specific."
)
raise ClanError(error)
# we have exactly one result at this point
result = results[0]
if var_id == result.id:
return result
msg = f"Did you mean: {result.id}"
raise ClanError(msg)
def get_command(
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:
sys.stdout.buffer.write(var.value)
else:
print(f"{var.id}: {var.printable_value}")
def register_get_parser(parser: argparse.ArgumentParser) -> None:
machines_arg = parser.add_argument(
"machine",
help="The machine to print vars for",
)
add_dynamic_completer(machines_arg, complete_machines)
parser.add_argument(
"var_id",
help="The var id to get the value for. Example: ssh-keys/pubkey",
)
parser.add_argument(
"--quiet",
"-q",
help="Only print the value of the var",
action="store_true",
)
parser.set_defaults(func=lambda args: get_command(**vars(args)))

View File

@@ -11,7 +11,7 @@ log = logging.getLogger(__name__)
# TODO get also secret facts # TODO get also secret facts
def get_all_vars(machine: Machine) -> list[Var]: def all_vars(machine: Machine) -> list[Var]:
public_vars_module = importlib.import_module(machine.public_vars_module) public_vars_module = importlib.import_module(machine.public_vars_module)
public_vars_store = public_vars_module.FactStore(machine=machine) public_vars_store = public_vars_module.FactStore(machine=machine)
secret_vars_module = importlib.import_module(machine.secret_vars_module) secret_vars_module = importlib.import_module(machine.secret_vars_module)
@@ -24,19 +24,19 @@ def stringify_vars(_vars: list[Var]) -> str:
def stringify_all_vars(machine: Machine) -> str: def stringify_all_vars(machine: Machine) -> str:
return stringify_vars(get_all_vars(machine)) return stringify_vars(all_vars(machine))
def get_command(args: argparse.Namespace) -> None: def list_command(args: argparse.Namespace) -> None:
machine = Machine(name=args.machine, flake=args.flake) machine = Machine(name=args.machine, flake=args.flake)
print(stringify_all_vars(machine)) print(stringify_all_vars(machine))
def register_list_parser(parser: argparse.ArgumentParser) -> None: def register_list_parser(parser: argparse.ArgumentParser) -> None:
machines_parser = parser.add_argument( machines_arg = parser.add_argument(
"machine", "machine",
help="The machine to print facts for", help="The machine to print vars for",
) )
add_dynamic_completer(machines_parser, complete_machines) add_dynamic_completer(machines_arg, complete_machines)
parser.set_defaults(func=get_command) parser.set_defaults(func=list_command)