vars: implement listing all vars
This commit is contained in:
50
pkgs/clan-cli/clan_cli/vars/_types.py
Normal file
50
pkgs/clan-cli/clan_cli/vars/_types.py
Normal file
@@ -0,0 +1,50 @@
|
||||
# !/usr/bin/env python3
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass
|
||||
|
||||
from clan_cli.machines.machines import Machine
|
||||
|
||||
|
||||
@dataclass
|
||||
class Var:
|
||||
store: "StoreBase"
|
||||
generator: str
|
||||
name: str
|
||||
secret: bool
|
||||
shared: bool
|
||||
deployed: bool
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.secret:
|
||||
return f"{self.generator}/{self.name}: ********"
|
||||
return f"{self.generator}/{self.name}: {self.store.get(self.generator, self.name, self.shared).decode()}"
|
||||
|
||||
|
||||
class StoreBase(ABC):
|
||||
def __init__(self, machine: Machine) -> None:
|
||||
self.machine = machine
|
||||
|
||||
@abstractmethod
|
||||
def exists(self, service: str, name: str, shared: bool = False) -> bool:
|
||||
pass
|
||||
|
||||
# get a single fact
|
||||
@abstractmethod
|
||||
def get(self, service: str, name: str, shared: bool = False) -> bytes:
|
||||
pass
|
||||
|
||||
def get_all(self) -> list[Var]:
|
||||
all_vars = []
|
||||
for gen_name, generator in self.machine.vars_generators.items():
|
||||
for f_name, file in generator["files"].items():
|
||||
all_vars.append(
|
||||
Var(
|
||||
store=self,
|
||||
generator=gen_name,
|
||||
name=f_name,
|
||||
secret=file["secret"],
|
||||
shared=generator["share"],
|
||||
deployed=file["deploy"],
|
||||
)
|
||||
)
|
||||
return all_vars
|
||||
@@ -1,40 +1,28 @@
|
||||
import argparse
|
||||
import importlib
|
||||
import json
|
||||
import logging
|
||||
|
||||
from ..completions import add_dynamic_completer, complete_machines
|
||||
from ..machines.machines import Machine
|
||||
from ._types import Var
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# TODO get also secret facts
|
||||
def get_all_facts(machine: Machine) -> dict:
|
||||
public_facts_module = importlib.import_module(machine.public_facts_module)
|
||||
public_facts_store = public_facts_module.FactStore(machine=machine)
|
||||
|
||||
# for service in machine.secrets_data:
|
||||
# facts[service] = {}
|
||||
# for fact in machine.secrets_data[service]["facts"]:
|
||||
# fact_content = fact_store.get(service, fact)
|
||||
# if fact_content:
|
||||
# facts[service][fact] = fact_content.decode()
|
||||
# else:
|
||||
# log.error(f"Fact {fact} for service {service} is missing")
|
||||
return public_facts_store.get_all()
|
||||
def get_all_vars(machine: Machine) -> list[Var]:
|
||||
public_vars_module = importlib.import_module(machine.public_vars_module)
|
||||
public_vars_store = public_vars_module.FactStore(machine=machine)
|
||||
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()
|
||||
|
||||
|
||||
def get_command(args: argparse.Namespace) -> None:
|
||||
machine = Machine(name=args.machine, flake=args.flake)
|
||||
|
||||
# the raw_facts are bytestrings making them not json serializable
|
||||
raw_facts = get_all_facts(machine)
|
||||
facts = dict()
|
||||
for key in raw_facts["TODO"]:
|
||||
facts[key] = raw_facts["TODO"][key].decode("utf8")
|
||||
|
||||
print(json.dumps(facts, indent=4))
|
||||
for var in get_all_vars(machine):
|
||||
print(var)
|
||||
|
||||
|
||||
def register_list_parser(parser: argparse.ArgumentParser) -> None:
|
||||
|
||||
@@ -1,30 +1,12 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from abc import abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
from clan_cli.machines.machines import Machine
|
||||
from clan_cli.vars._types import StoreBase
|
||||
|
||||
|
||||
class FactStoreBase(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self, machine: Machine) -> None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def exists(self, service: str, name: str, shared: bool = False) -> bool:
|
||||
pass
|
||||
|
||||
class FactStoreBase(StoreBase):
|
||||
@abstractmethod
|
||||
def set(
|
||||
self, service: str, name: str, value: bytes, shared: bool = False
|
||||
) -> Path | None:
|
||||
pass
|
||||
|
||||
# get a single fact
|
||||
@abstractmethod
|
||||
def get(self, service: str, name: str, shared: bool = False) -> bytes:
|
||||
pass
|
||||
|
||||
# get all facts
|
||||
@abstractmethod
|
||||
def get_all(self) -> dict[str, dict[str, bytes]]:
|
||||
pass
|
||||
|
||||
@@ -41,15 +41,3 @@ class FactStore(FactStoreBase):
|
||||
# get a single fact
|
||||
def get(self, generator_name: str, name: str, shared: bool = False) -> bytes:
|
||||
return self._var_path(generator_name, name, shared).read_bytes()
|
||||
|
||||
# get all public vars
|
||||
def get_all(self) -> dict[str, dict[str, bytes]]:
|
||||
facts: dict[str, dict[str, bytes]] = {}
|
||||
facts["TODO"] = {}
|
||||
if self.per_machine_folder.exists():
|
||||
for fact_path in self.per_machine_folder.iterdir():
|
||||
facts["TODO"][fact_path.name] = fact_path.read_bytes()
|
||||
if self.shared_folder.exists():
|
||||
for fact_path in self.shared_folder.iterdir():
|
||||
facts["TODO"][fact_path.name] = fact_path.read_bytes()
|
||||
return facts
|
||||
|
||||
@@ -35,14 +35,3 @@ class FactStore(FactStoreBase):
|
||||
if fact_path.exists():
|
||||
return fact_path.read_bytes()
|
||||
raise ClanError(f"Fact {name} for service {service} not found")
|
||||
|
||||
# get all facts
|
||||
def get_all(self) -> dict[str, dict[str, bytes]]:
|
||||
facts: dict[str, dict[str, bytes]] = {}
|
||||
if self.dir.exists():
|
||||
for service in self.dir.iterdir():
|
||||
facts[service.name] = {}
|
||||
for fact in service.iterdir():
|
||||
facts[service.name][fact.name] = fact.read_bytes()
|
||||
|
||||
return facts
|
||||
|
||||
@@ -1,14 +1,10 @@
|
||||
from abc import ABC, abstractmethod
|
||||
from abc import abstractmethod
|
||||
from pathlib import Path
|
||||
|
||||
from clan_cli.machines.machines import Machine
|
||||
from clan_cli.vars._types import StoreBase
|
||||
|
||||
|
||||
class SecretStoreBase(ABC):
|
||||
@abstractmethod
|
||||
def __init__(self, machine: Machine) -> None:
|
||||
pass
|
||||
|
||||
class SecretStoreBase(StoreBase):
|
||||
@abstractmethod
|
||||
def set(
|
||||
self,
|
||||
@@ -21,14 +17,6 @@ class SecretStoreBase(ABC):
|
||||
) -> Path | None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get(self, service: str, name: str, shared: bool = False) -> bytes:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def exists(self, service: str, name: str, shared: bool = False) -> bool:
|
||||
pass
|
||||
|
||||
def update_check(self) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
@@ -114,7 +114,10 @@ def test_all_dataclasses() -> None:
|
||||
|
||||
# Excludes:
|
||||
# - API includes Type Generic wrappers, that are not known in the init file.
|
||||
excludes = ["api/__init__.py"]
|
||||
excludes = [
|
||||
"api/__init__.py",
|
||||
"vars/_types.py",
|
||||
]
|
||||
|
||||
cli_path = Path("clan_cli").resolve()
|
||||
dataclasses = find_dataclasses_in_directory(cli_path, excludes)
|
||||
|
||||
Reference in New Issue
Block a user