add factsStore modules

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

@@ -0,0 +1,21 @@
+# !/usr/bin/env python3
+import argparse
+
+from .check import register_check_parser
+from .list import register_list_parser
+
+
+# takes a (sub)parser and configures it
+def register_parser(parser: argparse.ArgumentParser) -> None:
+    subparser = parser.add_subparsers(
+        title="command",
+        description="the command to run",
+        help="the command to run",
+        required=True,
+    )
+
+    check_parser = subparser.add_parser("check", help="check if facts are up to date")
+    register_check_parser(check_parser)
+
+    list_parser = subparser.add_parser("list", help="list all facts")
+    register_list_parser(list_parser)

pkgs/clan-cli/clan_cli/facts/check.py

@@ -0,0 +1,37 @@
+import argparse
+import importlib
+import logging
+
+from ..machines.machines import Machine
+
+log = logging.getLogger(__name__)
+
+
+def check_facts(machine: Machine) -> bool:
+    facts_module = importlib.import_module(machine.facts_module)
+    fact_store = facts_module.FactStore(machine=machine)
+
+    missing_facts = []
+    for service in machine.secrets_data:
+        for fact in machine.secrets_data[service]["facts"]:
+            if not fact_store.get(service, fact):
+                log.info(f"Fact {fact} for service {service} is missing")
+                missing_facts.append((service, fact))
+
+    if missing_facts:
+        return False
+    return True
+
+
+def check_command(args: argparse.Namespace) -> None:
+    machine = Machine(name=args.machine, flake=args.flake)
+    if check_facts(machine):
+        print("All facts are present")
+
+
+def register_check_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine",
+        help="The machine to check facts for",
+    )
+    parser.set_defaults(func=check_command)

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

@@ -0,0 +1,36 @@
+import json
+import argparse
+import importlib
+import logging
+
+from ..machines.machines import Machine
+
+log = logging.getLogger(__name__)
+
+
+def get_all_facts(machine: Machine) -> dict:
+    facts_module = importlib.import_module(machine.facts_module)
+    fact_store = 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 fact_store.get_all()
+
+
+def get_command(args: argparse.Namespace) -> None:
+    machine = Machine(name=args.machine, flake=args.flake)
+    print(json.dumps(get_all_facts(machine), indent=4))
+
+
+def register_list_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine",
+        help="The machine to print facts for",
+    )
+    parser.set_defaults(func=get_command)

pkgs/clan-cli/clan_cli/facts/modules/in_repo.py

@@ -0,0 +1,42 @@
+from pathlib import Path
+
+from clan_cli.errors import ClanError
+from clan_cli.machines.machines import Machine
+
+
+class FactStore:
+    def __init__(self, machine: Machine) -> None:
+        self.machine = machine
+
+    def set(self, _service: str, name: str, value: bytes) -> Path | None:
+        if isinstance(self.machine.flake, Path):
+            fact_path = (
+                self.machine.flake / "machines" / self.machine.name / "facts" / name
+            )
+            fact_path.parent.mkdir(parents=True, exist_ok=True)
+            fact_path.touch()
+            fact_path.write_bytes(value)
+            return fact_path
+        else:
+            raise ClanError(
+                f"in_flake fact storage is only supported for local flakes: {self.machine.flake}"
+            )
+
+    def exists(self, _service: str, name: str) -> bool:
+        fact_path = self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
+        return fact_path.exists()
+
+    # get a single fact
+    def get(self, _service: str, name: str) -> bytes:
+        fact_path = self.machine.flake_dir / "machines" / self.machine.name / "facts" / name
+        return fact_path.read_bytes()
+
+    # get all facts
+    def get_all(self) -> dict[str, dict[str, bytes]]:
+        facts_folder = self.machine.flake_dir / "machines" / self.machine.name / "facts"
+        facts: dict[str, dict[str, bytes]] = {}
+        facts["TODO"] = {}
+        if facts_folder.exists():
+            for fact_path in facts_folder.iterdir():
+                facts["TODO"][fact_path.name] = fact_path.read_bytes()
+        return facts