flupdt/__init__.py

@@ -0,0 +1 @@
+"""Tools to build, evaluate, and parse a nix flake."""

flupdt/cli.py

@@ -1,9 +1,16 @@
+"""Manages the CLI component of the tool."""
+
 import argparse
 
 
 def parse_inputs() -> argparse.Namespace:
+    """Parse inputs from argparse.
+
+    :returns the argparse Namespace to be evaluated
+    """
     parser = argparse.ArgumentParser()
     parser.add_argument("flake_path", metavar="flake-path", help="path to flake to evaluate")
-    parser.add_argument("--keep-hydra", action="store_true", help="allow evaluating Hydra jobs")
-    args = parser.parse_args()
-    return args
+    parser.add_argument("--keep-hydra", action="store_true", help="retain Hydra jobs")
+    parser.add_argument("--build", action="store_true", help="allow building Hydra jobs")
+    parser.add_argument("--evaluate", action="store_true", help="allow evaluating Hydra jobs")
+    return parser.parse_args()

flupdt/common.py

@@ -1,9 +1,11 @@
-"""common."""
+"""Common utilities."""
 
 import itertools
 import logging
 import sys
-from subprocess import Popen,PIPE
+from collections.abc import Iterable
+from subprocess import PIPE, Popen
+from types import FunctionType
 
 
 def configure_logger(level: str = "INFO") -> None:
@@ -20,7 +22,7 @@ def configure_logger(level: str = "INFO") -> None:
     )
 
 
-def partition(predicate, iterable):
+def partition(predicate: FunctionType, iterable: Iterable) -> tuple[Iterable, Iterable]:
     """Partition entries into false entries and true entries.
 
     If *predicate* is slow, consider wrapping it with functools.lru_cache().

flupdt/flake_build.py

@@ -0,0 +1,31 @@
+"""Provides components to build nix components and process the result."""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from flupdt.common import bash_wrapper
+
+drv_re = re.compile(r".*(/nix/store/.*\.drv).*")
+
+
+def build_output(path: str, output: str) -> str | None:
+    """Builds a given output in a flake.
+
+    :param path: path to flake
+    :param output: flake output to be built
+    :returns the .drv path on success or None on failure
+    """
+    logging.info(f"build {output}")
+    out = bash_wrapper(f"nix build {path}#{output} -o {output}.nixoutput")
+    logging.debug("output")
+    logging.debug(out[0])
+    logging.debug("error")
+    logging.debug(out[1])
+    logging.debug("statuscode")
+    logging.debug(out[2])
+    if out[2] != 0:
+        logging.warning(f"output {output} did not build correctly")
+        return None
+    return ""

flupdt/flake_eval.py

@@ -1,14 +1,23 @@
+"""Provides components to evaluate nix components and process the result."""
+
 from __future__ import annotations
 
 import logging
-from typing import Optional
-from flupdt.common import bash_wrapper
 import re
 
+from flupdt.common import bash_wrapper
+
 drv_re = re.compile(r".*(/nix/store/.*\.drv).*")
 
 
-def evaluate_output(path: str, output: str) -> Optional[str]:
+def evaluate_output(path: str, output: str) -> str | None:
+    """Evaluates a given output in a flake.
+
+    :param path: path to flake
+    :param output: flake output to be evaluated
+    :returns the .drv path on success or None on failure
+    :raises RuntimeError: evaluation succeeds but no derivation is found
+    """
     logging.info(f"evaluating {output}")
     out = bash_wrapper(f"nix eval {path}#{output}")
     logging.debug(out[0])
@@ -17,7 +26,6 @@ def evaluate_output(path: str, output: str) -> Optional[str]:
     if out[2] != 0:
         logging.warning(f"output {output} did not evaluate correctly")
         return None
-    else:
     drv_match = drv_re.match(out[0])
     if drv_match is None:
         out_msg = "derivation succeeded but output derivation does not contain a derivation"

flupdt/flake_show.py

@@ -1,11 +1,10 @@
-#!/usr/bin/env python3
+"""Utility to extract flake output info using nix flake (show|check)."""
 
 import json
 import logging
 import re
 import shutil
 import typing
-from subprocess import Popen
 
 from flupdt.common import bash_wrapper
 
@@ -16,6 +15,12 @@ output_regexes = [
 
 
 def traverse_json_base(json_dict: dict[str, typing.Any], path: list[str]) -> list[str]:
+    """Crawls through the flake outputs to get nixos-configuration and derivation types.
+
+    :param json_dict: dict of flake outputs to check
+    :param path: a list of outputs constructed so far
+    :returns the output path list, plus any new paths found
+    """
     final_paths = []
     for key, value in json_dict.items():
         if isinstance(value, dict):
@@ -32,10 +37,21 @@ def traverse_json_base(json_dict: dict[str, typing.Any], path: list[str]) -> lis
 
 
 def traverse_json(json_dict: dict) -> list[str]:
+    """Crawls through the flake outputs to get nixos-configuration and derivation types.
+
+    :param json_dict: dict of flake outputs to check
+    :returns a list of outputs that can be evaluated
+    """
     return traverse_json_base(json_dict, [])
 
 
 def get_derivations_from_check(nix_path: str, path_to_flake: str) -> list[str]:
+    """Gets all derivations in a flake, using check instead of show.
+
+    :param nix_path: path to nix binary
+    :param path_to_flake: path to flake to be checked
+    :returns a list of all valid derivations in the flake
+    """
     flake_check = bash_wrapper(f"{nix_path} flake check --verbose --keep-going", path=path_to_flake)
     if flake_check[2] != 0:
         logging.warning(
@@ -55,10 +71,17 @@ def get_derivations_from_check(nix_path: str, path_to_flake: str) -> list[str]:
 
 
 def get_derivations(path_to_flake: str) -> list[str]:
+    """Gets all derivations present in a flake.
+
+    :param path_to_flake: path to flake to be checked
+    :returns a list of all valid derivations in the flake
+    :raises RuntimeError: fails if nix is not present in the PATH
+    """
     nix_path = shutil.which("nix")
     derivations = []
     if nix_path is None:
-        raise RuntimeError("nix is not available in the PATH, please verify that it is installed")
+        status_msg = "nix is not available in the PATH, please verify that it is installed"
+        raise RuntimeError(status_msg)
     flake_show = bash_wrapper(f"{nix_path} flake show --json", path=path_to_flake)
     if flake_show[2] != 0:
         logging.error("flake show returned non-zero exit code")

flupdt/main.py

@@ -1,10 +1,30 @@
 #!/usr/bin/env python3
 
-from flupdt.flake_show import get_derivations
-from flupdt.cli import parse_inputs
-from flupdt.flake_eval import evaluate_output
-from flupdt.common import configure_logger, partition
+"""Default processing of flake outputs for evaluating flake updates."""
+
 import logging
+from argparse import Namespace
+
+from flupdt.cli import parse_inputs
+from flupdt.common import configure_logger, partition
+from flupdt.flake_build import build_output
+from flupdt.flake_eval import evaluate_output
+from flupdt.flake_show import get_derivations
+
+
+def batch_eval(args: Namespace, flake_path: str, derivations: list[str]) -> None:
+    """Bulk run evaluations or builds on a derivation set.
+
+    :params args: argument namespace to check against
+    :params flake_path: path to flake to be evaluated
+    :params derivations: list of derivations to run against
+    :returns None
+    """
+    for d in derivations:
+        if args.evaluate:
+            evaluate_output(flake_path, d)
+        if args.build:
+            build_output(flake_path, d)
 
 
 def main() -> None:
@@ -13,23 +33,20 @@ def main() -> None:
     :returns: None
 
     """
-    configure_logger(logging.DEBUG)
+    configure_logger("DEBUG")
     args = parse_inputs()
     flake_path = args.flake_path
     derivations, hydra_jobs = partition(
         lambda s: s.startswith("hydraJobs"), get_derivations(flake_path)
     )
+    derivations, hydra_jobs = list(derivations), list(hydra_jobs)
     logging.info(f"derivations: {list(derivations)}")
-    for d in derivations:
-        evaluate_output(flake_path, d)
+    batch_eval(args, flake_path, derivations)
 
     if not args.keep_hydra:
         logging.info("--keep-hydra flag is not specified, removing Hydra jobs")
     else:
-        hydra_jobs = list(hydra_jobs)
-        logging.info(f"hydraJobs: {hydra_jobs}")
-        for d in hydra_jobs:
-            evaluate_output(flake_path, d)
+        batch_eval(args, flake_path, hydra_jobs)
 
 
 if __name__ == "__main__":