From 0db5abf56abc350a3fd6c88c8a7e7912e9256688 Mon Sep 17 00:00:00 2001 From: Qubasa Date: Thu, 30 Jan 2025 16:24:50 +0700 Subject: [PATCH] clan-cli: Make clan flakes create discover templates from inputs. Add clan flakes list command --- flake.nix | 6 + pkgs/clan-cli/clan_cli/clan/__init__.py | 3 + pkgs/clan-cli/clan_cli/clan/create.py | 164 +++++++++------ pkgs/clan-cli/clan_cli/clan/list.py | 24 +++ pkgs/clan-cli/clan_cli/machines/list.py | 4 +- pkgs/clan-cli/clan_cli/templates.py | 259 ++++++++++++++++++++++++ pkgs/clan-cli/default.nix | 5 + templates/default.nix | 39 ++++ templates/flake-module.nix | 8 +- 9 files changed, 438 insertions(+), 74 deletions(-) create mode 100644 pkgs/clan-cli/clan_cli/clan/list.py create mode 100644 pkgs/clan-cli/clan_cli/templates.py create mode 100644 templates/default.nix diff --git a/flake.nix b/flake.nix index 9fe0b3efc..7e6077210 100644 --- a/flake.nix +++ b/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 { }) diff --git a/pkgs/clan-cli/clan_cli/clan/__init__.py b/pkgs/clan-cli/clan_cli/clan/__init__.py index 41c8ae718..d55091a94 100644 --- a/pkgs/clan-cli/clan_cli/clan/__init__.py +++ b/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) diff --git a/pkgs/clan-cli/clan_cli/clan/create.py b/pkgs/clan-cli/clan_cli/clan/create.py index 6bdd52271..fe2446e0f 100644 --- a/pkgs/clan-cli/clan_cli/clan/create.py +++ b/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 + flake_update: CmdOut | None = None 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 - # URL to the template to use. Defaults to the "minimal" template - template: str = "minimal" + dest: Path + template_name: str + 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: - directory = Path(options.directory).resolve() - template_url = f"{clan_templates(TemplateType.CLAN)}#{options.template}" - if not directory.exists(): - directory.mkdir() - else: - # Directory already exists - # Check if it is empty - # 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.", +def create_clan(opts: CreateOptions) -> CreateClanResponse: + dest = opts.dest.resolve() + + template = get_template( + TemplateName(opts.template_name), + "clan", + input_prio=opts.input_prio, + clan_dir=opts.src_flake, + ) + log.info(f"Found template '{template.name}' in '{template.input_variant}'") + src = Path(template.src["path"]) + + if dest.exists(): + dest /= src.name + + if dest.exists(): + msg = f"Destination directory {dest} already exists" + raise ClanError(msg) + + if not src.exists(): + 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) + + copy_from_nixstore(src, dest) + + response = CreateClanResponse() + + 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(dest, "config", "user.name"), RunOpts(check=False) + ) + response.git_config_username = None + if has_username.returncode != 0: + response.git_config_username = run( + git_command(dest, "config", "user.name", "clan-tool") ) - command = nix_command( - [ - "flake", - "init", - "-t", - template_url, - ] - ) - flake_init = run(command, RunOpts(cwd=directory)) + has_username = run( + git_command(dest, "config", "user.email"), RunOpts(check=False) + ) + if has_username.returncode != 0: + response.git_config_email = run( + git_command(dest, "config", "user.email", "clan@example.com") + ) flake_update = run( - nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=directory) + nix_shell(["nixpkgs#nix"], ["nix", "flake", "update"]), RunOpts(cwd=dest) ) + response.flake_update = flake_update - if options.initial: - init_inventory(str(options.directory), init=options.initial) - - response = CreateClanResponse( - flake_init=flake_init, - flake_update=flake_update, - ) - if not options.setup_git: - return response - - response.git_init = run(git_command(directory, "init")) - response.git_add = run(git_command(directory, "add", ".")) - - # check if username is set - has_username = run( - git_command(directory, "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") - ) - - has_username = run( - git_command(directory, "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") - ) + 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, - template=args.template, + input_prio=input_prio, + dest=args.path, + template_name=args.template, setup_git=not args.no_git, + src_flake=args.flake, ) ) diff --git a/pkgs/clan-cli/clan_cli/clan/list.py b/pkgs/clan-cli/clan_cli/clan/list.py new file mode 100644 index 000000000..7594a6311 --- /dev/null +++ b/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) diff --git a/pkgs/clan-cli/clan_cli/machines/list.py b/pkgs/clan-cli/clan_cli/machines/list.py index 5a18e2c20..f81d41755 100644 --- a/pkgs/clan-cli/clan_cli/machines/list.py +++ b/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(): diff --git a/pkgs/clan-cli/clan_cli/templates.py b/pkgs/clan-cli/clan_cli/templates.py new file mode 100644 index 000000000..5b482fc3b --- /dev/null +++ b/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 + ) diff --git a/pkgs/clan-cli/default.nix b/pkgs/clan-cli/default.nix index 9a5feeff5..e34b9f476 100644 --- a/pkgs/clan-cli/default.nix +++ b/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)) diff --git a/templates/default.nix b/templates/default.nix new file mode 100644 index 000000000..2cc6adeb8 --- /dev/null +++ b/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; + }; + }; +} diff --git a/templates/flake-module.nix b/templates/flake-module.nix index 639d800c1..a48b38ce3 100644 --- a/templates/flake-module.nix +++ b/templates/flake-module.nix @@ -1,9 +1,9 @@ { self, inputs, ... }: { - flake = (import ./flake.nix).outputs { } // { - checks.x86_64-linux.template-minimal = + flake = { + 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; }; };