clan-cli: Make clan flakes create discover templates from inputs. Add clan flakes list command

flake.nix

@@ -43,6 +43,11 @@
           meta.name = "clan-core";
           inherit self;
         };
+
+        flake = {
+          clan.templates = import ./templates { };
+        };
+
         systems = import systems;
         imports =
           # only importing existing paths allows to minimize the flake for test
@@ -59,6 +64,7 @@
             ./nixosModules/flake-module.nix
             ./pkgs/flake-module.nix
             ./templates/flake-module.nix
+            ./new-templates/flake-module.nix
           ]
           ++ [
             (if pathExists ./flakeModules/clan.nix then import ./flakeModules/clan.nix inputs.self else { })

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

@@ -4,6 +4,7 @@ import argparse
 from clan_cli.clan.inspect import register_inspect_parser
 
 from .create import register_create_parser
+from .list import register_list_parser
 
 
 # takes a (sub)parser and configures it
@@ -18,3 +19,5 @@ def register_parser(parser: argparse.ArgumentParser) -> None:
     register_create_parser(create_parser)
     inspect_parser = subparser.add_parser("inspect", help="Inspect a clan ")
     register_inspect_parser(inspect_parser)
+    list_parser = subparser.add_parser("list", help="List clan templates")
+    register_list_parser(list_parser)

pkgs/clan-cli/clan_cli/clan/create.py

@@ -1,21 +1,28 @@
 # !/usr/bin/env python3
 import argparse
-import os
+import logging
 from dataclasses import dataclass
 from pathlib import Path
 
 from clan_cli.api import API
+from clan_cli.clan_uri import FlakeId
 from clan_cli.cmd import CmdOut, RunOpts, run
-from clan_cli.dirs import TemplateType, clan_templates
 from clan_cli.errors import ClanError
 from clan_cli.inventory import Inventory, init_inventory
-from clan_cli.nix import nix_command, nix_shell
+from clan_cli.nix import nix_shell
+from clan_cli.templates import (
+    InputPrio,
+    TemplateName,
+    copy_from_nixstore,
+    get_template,
+)
+
+log = logging.getLogger(__name__)
 
 
 @dataclass
 class CreateClanResponse:
-    flake_init: CmdOut
+    flake_update: CmdOut | None = None
-    flake_update: CmdOut
     git_init: CmdOut | None = None
     git_add: CmdOut | None = None
     git_config_username: CmdOut | None = None
@@ -24,9 +31,10 @@ class CreateClanResponse:
 
 @dataclass
 class CreateOptions:
-    directory: Path | str
+    dest: Path
-    # URL to the template to use. Defaults to the "minimal" template
+    template_name: str
-    template: str = "minimal"
+    src_flake: FlakeId | None = None
+    input_prio: InputPrio | None = None
     setup_git: bool = True
     initial: Inventory | None = None
 
@@ -36,76 +44,88 @@ def git_command(directory: Path, *args: str) -> list[str]:
 
 
 @API.register
-def create_clan(options: CreateOptions) -> CreateClanResponse:
+def create_clan(opts: CreateOptions) -> CreateClanResponse:
-    directory = Path(options.directory).resolve()
+    dest = opts.dest.resolve()
-    template_url = f"{clan_templates(TemplateType.CLAN)}#{options.template}"
+
-    if not directory.exists():
+    template = get_template(
-        directory.mkdir()
+        TemplateName(opts.template_name),
-    else:
+        "clan",
-        # Directory already exists
+        input_prio=opts.input_prio,
-        # Check if it is empty
+        clan_dir=opts.src_flake,
-        # Throw error otherwise
-        dir_content = os.listdir(directory)
-        if len(dir_content) != 0:
-            raise ClanError(
-                location=f"{directory.resolve()}",
-                msg="Cannot create clan",
-                description="Directory already exists and is not empty.",
     )
+    log.info(f"Found template '{template.name}' in '{template.input_variant}'")
+    src = Path(template.src["path"])
 
-    command = nix_command(
+    if dest.exists():
-        [
+        dest /= src.name
-            "flake",
-            "init",
-            "-t",
-            template_url,
-        ]
-    )
-    flake_init = run(command, RunOpts(cwd=directory))
 
-    flake_update = run(
+    if dest.exists():
-        nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=directory)
+        msg = f"Destination directory {dest} already exists"
-    )
+        raise ClanError(msg)
 
-    if options.initial:
+    if not src.exists():
-        init_inventory(str(options.directory), init=options.initial)
+        msg = f"Template {template} does not exist"
+        raise ClanError(msg)
+    if not src.is_dir():
+        msg = f"Template {template} is not a directory"
+        raise ClanError(msg)
 
-    response = CreateClanResponse(
+    copy_from_nixstore(src, dest)
-        flake_init=flake_init,
-        flake_update=flake_update,
-    )
-    if not options.setup_git:
-        return response
 
-    response.git_init = run(git_command(directory, "init"))
+    response = CreateClanResponse()
-    response.git_add = run(git_command(directory, "add", "."))
+
+    if opts.setup_git:
+        response.git_init = run(git_command(dest, "init"))
+        response.git_add = run(git_command(dest, "add", "."))
 
         # check if username is set
         has_username = run(
-        git_command(directory, "config", "user.name"), RunOpts(check=False)
+            git_command(dest, "config", "user.name"), RunOpts(check=False)
         )
         response.git_config_username = None
         if has_username.returncode != 0:
             response.git_config_username = run(
-            git_command(directory, "config", "user.name", "clan-tool")
+                git_command(dest, "config", "user.name", "clan-tool")
             )
 
         has_username = run(
-        git_command(directory, "config", "user.email"), RunOpts(check=False)
+            git_command(dest, "config", "user.email"), RunOpts(check=False)
         )
         if has_username.returncode != 0:
             response.git_config_email = run(
-            git_command(directory, "config", "user.email", "clan@example.com")
+                git_command(dest, "config", "user.email", "clan@example.com")
             )
 
+    flake_update = run(
+        nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=dest)
+    )
+    response.flake_update = flake_update
+
+    if opts.initial:
+        init_inventory(str(opts.dest), init=opts.initial)
+
     return response
 
 
 def register_create_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "--input",
+        type=str,
+        help="Flake input name to use as template source",
+        action="append",
+        default=[],
+    )
+
+    parser.add_argument(
+        "--no-self-prio",
+        help="Do not prioritize 'self' input",
+        action="store_true",
+        default=False,
+    )
+
     parser.add_argument(
         "--template",
         type=str,
-        choices=["default", "minimal", "flake-parts", "minimal-flake-parts"],
         help="Clan template name",
         default="default",
     )
@@ -118,15 +138,25 @@ def register_create_parser(parser: argparse.ArgumentParser) -> None:
     )
 
     parser.add_argument(
-        "path", type=Path, help="Path to the clan directory", default=Path()
+        "path",
+        type=Path,
+        help="Path where to write the clan template to",
+        default=Path(),
     )
 
     def create_flake_command(args: argparse.Namespace) -> None:
+        if args.no_self_prio:
+            input_prio = InputPrio.try_inputs(tuple(args.input))
+        else:
+            input_prio = InputPrio.try_self_then_inputs(tuple(args.input))
+
         create_clan(
             CreateOptions(
-                directory=args.path,
+                input_prio=input_prio,
-                template=args.template,
+                dest=args.path,
+                template_name=args.template,
                 setup_git=not args.no_git,
+                src_flake=args.flake,
             )
         )
 

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

@@ -0,0 +1,24 @@
+import argparse
+import logging
+
+from clan_cli.templates import list_templates
+
+log = logging.getLogger(__name__)
+
+
+def list_command(args: argparse.Namespace) -> None:
+    template_list = list_templates("clan", args.flake)
+
+    print("Available local templates:")
+    for name, template in template_list.self.items():
+        print(f"  {name}: {template['description']}")
+
+    print("Available templates from inputs:")
+    for input_name, input_templates in template_list.inputs.items():
+        print(f"  {input_name}:")
+        for name, template in input_templates.items():
+            print(f"    {name}: {template['description']}")
+
+
+def register_list_parser(parser: argparse.ArgumentParser) -> None:
+    parser.set_defaults(func=list_command)

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

@@ -1,6 +1,7 @@
 import argparse
 import json
 import logging
+import re
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Literal
@@ -45,9 +46,6 @@ class MachineDetails:
     disk_schema: MachineDiskMatter | None = None
 
 
-import re
-
-
 def extract_header(c: str) -> str:
     header_lines = []
     for line in c.splitlines():

pkgs/clan-cli/clan_cli/templates.py

@@ -0,0 +1,259 @@
+import json
+import logging
+import os
+import shutil
+import stat
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Literal, NewType, TypedDict
+
+from clan_cli.clan_uri import FlakeId  # Custom FlakeId type for Nix Flakes
+from clan_cli.cmd import run  # Command execution utility
+from clan_cli.errors import ClanError  # Custom exception for Clan errors
+from clan_cli.nix import nix_eval  # Helper for Nix evaluation scripts
+
+# Configure the logging module for debugging
+log = logging.getLogger(__name__)
+
+# Define custom types for better type annotations
+
+InputName = NewType("InputName", str)
+
+
+@dataclass
+class InputVariant:
+    input_name: InputName | None
+
+    def is_self(self) -> bool:
+        return self.input_name is None
+
+    def __str__(self) -> str:
+        return self.input_name or "self"
+
+
+TemplateName = NewType("TemplateName", str)  # Represents the name of a template
+TemplateType = Literal[
+    "clan", "disko", "machine"
+]  # Literal to restrict template type to specific values
+ModuleName = NewType("ModuleName", str)  # Represents a module name in Clan exports
+
+
+# TypedDict for ClanModule with required structure
+class ClanModule(TypedDict):
+    description: str  # Description of the module
+    path: str  # Filepath of the module
+
+
+# TypedDict for a Template with required structure
+class Template(TypedDict):
+    description: str  # Template description
+    path: str  # Template path on disk
+
+
+# TypedDict for the structure of templates organized by type
+class TemplateTypeDict(TypedDict):
+    disko: dict[TemplateName, Template]  # Templates under "disko" type
+    clan: dict[TemplateName, Template]  # Templates under "clan" type
+    machine: dict[TemplateName, Template]  # Templates under "machine" type
+
+
+# TypedDict for a Clan attribute set (attrset) with templates and modules
+class ClanAttrset(TypedDict):
+    templates: TemplateTypeDict  # Organized templates by type
+    modules: dict[ModuleName, ClanModule]  # Dictionary of modules by module name
+
+
+# TypedDict to represent exported Clan attributes
+class ClanExports(TypedDict):
+    inputs: dict[
+        InputName, ClanAttrset
+    ]  # Input names map to their corresponding attrsets
+    self: ClanAttrset  # The attribute set for the flake itself
+
+
+# Helper function to get Clan exports (Nix flake outputs)
+def get_clan_exports(clan_dir: FlakeId | None = None) -> ClanExports:
+    # Check if the clan directory is provided, otherwise use the environment variable
+    if not clan_dir:
+        clan_core_path = os.environ.get("CLAN_CORE_PATH")
+        if not clan_core_path:
+            msg = "Environment var CLAN_CORE_PATH is not set, this shouldn't happen"
+            raise ClanError(msg)
+        # Use the clan core path from the environment variable
+        clan_dir = FlakeId(clan_core_path)
+
+    log.debug(f"Evaluating flake {clan_dir} for Clan attrsets")
+
+    # Nix evaluation script to compute the relevant exports
+    eval_script = f"""
+        let
+            self = builtins.getFlake "{clan_dir}";
+            lib = self.inputs.nixpkgs.lib;
+            inputsWithClan = lib.mapAttrs (
+                _name: value: value.clan
+            ) (lib.filterAttrs(_name: value: value ? "clan") self.inputs);
+        in
+            {{ inputs = inputsWithClan; self = self.clan or {{}}; }}
+    """
+
+    # Evaluate the Nix expression and run the command
+    cmd = nix_eval(
+        [
+            "--json",  # Output the result as JSON
+            "--impure",  # Allow impure evaluations (env vars or system state)
+            "--expr",  # Evaluate the given Nix expression
+            eval_script,
+        ]
+    )
+    res = run(cmd).stdout  # Run the command and capture the JSON output
+    return json.loads(res)  # Parse and return as a Python dictionary
+
+
+# Dataclass to manage input prioritization for templates
+@dataclass
+class InputPrio:
+    input_names: tuple[str, ...]  # Tuple of input names (ordered priority list)
+    prioritize_self: bool = True  # Whether to prioritize "self" first
+
+    # Static factory methods for specific prioritization strategies
+
+    @staticmethod
+    def self_only() -> "InputPrio":
+        # Only consider "self" (no external inputs)
+        return InputPrio(prioritize_self=True, input_names=())
+
+    @staticmethod
+    def try_inputs(input_names: tuple[str, ...]) -> "InputPrio":
+        # Only consider the specified external inputs
+        return InputPrio(prioritize_self=False, input_names=input_names)
+
+    @staticmethod
+    def try_self_then_inputs(input_names: tuple[str, ...]) -> "InputPrio":
+        # Consider "self" first, then the specified external inputs
+        return InputPrio(prioritize_self=True, input_names=input_names)
+
+
+@dataclass
+class FoundTemplate:
+    input_variant: InputVariant
+    name: TemplateName
+    src: Template
+
+
+def copy_from_nixstore(src: Path, dest: Path) -> None:
+    if not src.is_dir():
+        msg = f"The source path '{src}' is not a directory."
+        raise ClanError(msg)
+
+    # Walk through the source directory
+    for root, _dirs, files in src.walk():
+        relative_path = Path(root).relative_to(src)
+        dest_dir = dest / relative_path
+
+        dest_dir.mkdir(exist_ok=True)
+        log.debug(f"Creating directory '{dest_dir}'")
+        # Set permissions for directories
+        dest_dir.chmod(stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
+
+        for file_name in files:
+            src_file = Path(root) / file_name
+            dest_file = dest_dir / file_name
+
+            # Copy the file
+            shutil.copy(src_file, dest_file)
+
+            dest_file.chmod(stat.S_IWRITE | stat.S_IREAD)
+
+
+@dataclass
+class TemplateList:
+    inputs: dict[InputName, dict[TemplateName, Template]] = field(default_factory=dict)
+    self: dict[TemplateName, Template] = field(default_factory=dict)
+
+
+def list_templates(
+    template_type: TemplateType, clan_dir: FlakeId | None = None
+) -> TemplateList:
+    clan_exports = get_clan_exports(clan_dir)
+    result = TemplateList()
+    fallback: ClanAttrset = {
+        "templates": {"disko": {}, "clan": {}, "machine": {}},
+        "modules": {},
+    }
+
+    clan_templates = (
+        clan_exports["self"]
+        .get("templates", fallback["templates"])
+        .get(template_type, {})
+    )
+    result.self = clan_templates
+    for input_name, _attrset in clan_exports["inputs"].items():
+        clan_templates = (
+            clan_exports["inputs"]
+            .get(input_name, fallback)["templates"]
+            .get(template_type, {})
+        )
+        result.inputs[input_name] = {}
+        for template_name, template in clan_templates.items():
+            result.inputs[input_name][template_name] = template
+
+    return result
+
+
+# Function to retrieve a specific template from Clan exports
+def get_template(
+    template_name: TemplateName,
+    template_type: TemplateType,
+    *,
+    input_prio: InputPrio | None = None,
+    clan_dir: FlakeId | None = None,
+) -> FoundTemplate:
+    log.info(f"Searching for template '{template_name}' of type '{template_type}'")
+
+    # Set default priority strategy if none is provided
+    if input_prio is None:
+        input_prio = InputPrio.try_self_then_inputs(("clan-core",))
+
+    # Helper function to search for a specific template within a dictionary of templates
+    def find_template(
+        template_name: TemplateName, templates: dict[TemplateName, Template]
+    ) -> Template | None:
+        if template_name in templates:
+            return templates[template_name]
+        return None
+
+    # Initialize variables for the search results
+    template: Template | None = None
+    input_name: InputName | None = None
+    template_list = list_templates(template_type, clan_dir)
+
+    # Step 1: Check "self" first, if prioritize_self is enabled
+    if input_prio.prioritize_self:
+        log.info(f"Checking 'self' for template '{template_name}'")
+        template = find_template(template_name, template_list.self)
+
+    # Step 2: Otherwise, check the external inputs if no match is found
+    if not template and input_prio.input_names:
+        log.info(f"Checking external inputs for template '{template_name}'")
+        for input_name_str in input_prio.input_names:
+            input_name = InputName(input_name_str)
+            log.debug(f"Checking input '{input_name}' for template '{template_name}'")
+
+            template = find_template(template_name, template_list.inputs[input_name])
+            if template:
+                log.debug(f"Found template '{template_name}' in input '{input_name}'")
+                break  # Stop searching once the template is found
+
+    # Step 3: Raise an error if the template wasn't found
+    if not template:
+        source = (
+            f"inputs.{input_name}.clan.templates.{template_type}"
+            if input_name  # Most recent "input_name"
+            else f"flake.clan.templates.{template_type}"
+        )
+        msg = f"Template '{template_name}' not in '{source}' in '{clan_dir}'"
+        raise ClanError(msg)
+
+    return FoundTemplate(
+        input_variant=InputVariant(input_name), src=template, name=template_name
+    )

pkgs/clan-cli/default.nix

@@ -102,6 +102,11 @@ pythonRuntime.pkgs.buildPythonApplication {
     ":"
     (lib.makeBinPath (lib.attrValues includedRuntimeDependenciesMap))
 
+    # We need this for templates to work
+    "--set"
+    "CLAN_CORE_PATH"
+    clan-core-path
+
     "--set"
     "CLAN_STATIC_PROGRAMS"
     (lib.concatStringsSep ":" (lib.attrNames includedRuntimeDependenciesMap))

templates/default.nix

@@ -0,0 +1,39 @@
+{ ... }:
+{
+  disko = {
+    single-disk = {
+      description = "A simple ext4 disk with a single partition";
+      path = ./disk/single-disk;
+    };
+  };
+
+  machine = {
+    flash-installer = {
+      description = "Initialize a new flash-installer machine";
+      path = ./clan/machineTemplates/machines/flash-installer;
+    };
+    new-machine = {
+      description = "Initialize a new machine";
+      path = ./clan/machineTemplates/machines/new-machine;
+    };
+  };
+
+  clan = {
+    default = {
+      description = "Initialize a new clan flake";
+      path = ./clan/new-clan;
+    };
+    minimal = {
+      description = "for clans managed via (G)UI";
+      path = ./clan/minimal;
+    };
+    flake-parts = {
+      description = "Flake-parts";
+      path = ./clan/flake-parts;
+    };
+    minimal-flake-parts = {
+      description = "Minimal flake-parts clan template";
+      path = ./clan/minimal-flake-parts;
+    };
+  };
+}

templates/flake-module.nix

@@ -1,9 +1,9 @@
 { self, inputs, ... }:
 {
-  flake = (import ./flake.nix).outputs { } // {
+  flake = {
-    checks.x86_64-linux.template-minimal =
+    checks.x86_64-linux.new-template-minimal =
       let
-        path = self.templates.minimal.path;
+        path = self.clan.templates.clan.minimal.path;
         initialized = inputs.nixpkgs.legacyPackages.x86_64-linux.runCommand "minimal-clan-flake" { } ''
           mkdir $out
           cp -r ${path}/* $out
@@ -30,7 +30,7 @@
       in
       {
         type = "derivation";
-        name = "minimal-clan-flake-check";
+        name = "new-minimal-clan-flake-check";
         inherit (evaled.nixosConfigurations.testmachine.config.system.build.toplevel) drvPath outPath;
       };
   };