Add: flake attribute support

nix/eval-machine-info.nix

@@ -1,6 +1,7 @@
 { system ? builtins.currentSystem
 , networkExprs
-, flakeUri ? null
+, flakeReference ? null
+, flakeAttribute
 , checkConfigurationOptions ? true
 , uuid
 , deploymentName
@@ -9,6 +10,39 @@
 }:
 
 let
+  /* Return an attribute from nested attribute sets.
+     Example:
+       x = { a = { b = 3; }; }
+       attrByPath ["a" "b"] x
+       => 3
+       x = { a = { b = 3; }; }
+       attrByPath ["a" "c"] x
+       error: can't found `a.c` in the set
+  */
+  attrByPath = attrPath: e:
+    let
+      attrByPathRec  = attrPath: e:
+        # if it's the end return the element it's self
+        if attrPath == [] then
+          {inherit e; found = true;}
+        else
+          let attr = builtins.head attrPath; in
+          if e ? ${attr} then
+            attrByPathRec (builtins.tail attrPath) e.${attr}
+          else
+            {e=null; found = false;};
+          result = attrByPathRec attrPath e;
+      in
+        # Check result
+        if result.found then
+          # If found return the value
+          result.e
+        else
+          # Not found, Create the searching path
+          let path = builtins.concatStringsSep "." attrPath; in
+            # Throw an exception
+            throw "can't found `${path}` in the set";
+
   call = x: if builtins.isFunction x then x args else x;
 
   # Copied from nixpkgs to avoid <nixpkgs> import
@@ -17,7 +51,16 @@ let
   zipAttrs = set: builtins.listToAttrs (
     map (name: { inherit name; value = builtins.catAttrs name set; }) (builtins.concatMap builtins.attrNames set));
 
-  flakeExpr = (builtins.getFlake flakeUri).outputs.nixopsConfigurations.default;
+  # the flake expresion
+  flakeExpr =
+    let
+      # Get the flake config
+      flake = builtins.getFlake flakeReference;
+      # get the chosen deployement.
+      deploy = attrByPath flakeAttribute flake.outputs.nixopsConfigurations;
+    in
+      # Return the deployement found.
+      deploy;
 
   networks =
     let
@@ -32,20 +75,20 @@ let
       };
     in
       map ({ key }: getNetworkFromExpr key) networkExprClosure
-      ++ optional (flakeUri != null)
-        ((call flakeExpr) // { _file = "<${flakeUri}>"; });
+      ++ optional (flakeReference != null)
+        ((call flakeExpr) // { _file = "<${flakeReference}>"; });
 
   network = zipAttrs networks;
 
   evalConfig =
-    if flakeUri != null
+    if flakeReference != null
     then
       if network ? nixpkgs
       then (builtins.head (network.nixpkgs)).lib.nixosSystem
       else throw "NixOps network must have a 'nixpkgs' attribute"
     else import (pkgs.path + "/nixos/lib/eval-config.nix");
 
-  pkgs = if flakeUri != null
+  pkgs = if flakeReference != null
     then
       if network ? nixpkgs
       then (builtins.head network.nixpkgs).legacyPackages.${system}
@@ -293,7 +336,7 @@ in rec {
   getNixOpsArgs = fs: lib.zipAttrs (lib.unique (lib.concatMap fileToArgs (getNixOpsExprs fs)));
 
   nixopsArguments =
-    if flakeUri == null then getNixOpsArgs networkExprs
-    else lib.listToAttrs (builtins.map (a: {name = a; value = [ flakeUri ];}) (lib.attrNames (builtins.functionArgs flakeExpr)));
+    if flakeReference == null then getNixOpsArgs networkExprs
+    else lib.listToAttrs (builtins.map (a: {name = a; value = [ flakeReference ];}) (lib.attrNames (builtins.functionArgs flakeExpr)));
 
 }

nixops/evaluation.py

@@ -1,7 +1,7 @@
 from nixops.nix_expr import RawValue, py2nix
 import subprocess
 import typing
-from typing import Optional, Mapping, Any, List, Dict, TextIO
+from typing import Optional, Mapping, Any, List, Dict, TextIO, Union
 import json
 from nixops.util import ImmutableValidatedObject
 from nixops.exceptions import NixError
@@ -51,7 +51,11 @@ class EvalResult(ImmutableValidatedObject):
 @dataclass
 class NetworkFile:
     network: str
-    is_flake: bool = False
+    attribute: Union[str, None] = None
+
+    @property
+    def is_flake(self) -> bool:
+        return self.attribute != None
 
 
 def get_expr_path() -> str:
@@ -120,7 +124,8 @@ def eval(
 
     if networkExpr.is_flake:
         argv.extend(["--allowed-uris", get_expr_path()])
-        argv.extend(["--argstr", "flakeUri", networkExpr.network])
+        argv.extend(["--argstr", "flakeReference", networkExpr.network])
+        argv.extend(["--arg", "flakeAttribute", networkExpr.attribute or "null"])
 
     try:
         ret = subprocess.check_output(argv, stderr=stderr, text=True)

nixops/nix_expr.py

@@ -1,6 +1,7 @@
 import functools
+import re
 import string
-from typing import Optional, Any, List, Union, Dict
+from typing import Optional, Any, List, Tuple, Union, Dict
 from textwrap import dedent
 
 __all__ = ["py2nix", "nix2py", "nixmerge", "expand_dict", "RawValue", "Function"]
@@ -358,3 +359,61 @@ def nix2py(source: str) -> MultiLineRawValue:
     which are used as-is and only indentation will take place.
     """
     return MultiLineRawValue(dedent(source).strip().splitlines())
+
+
+# Regex to match nix string
+LITERAL_STRING_REGEX: str = r"\"(?:[^\"\\]|\\.)*\""
+# Regex to match nix string
+KEYWORD_REGEX: str = r"[a-zA-Z\_][a-zA-Z0-9\_\'\-]*"
+
+
+def _extract_key(path: str) -> Tuple[str, str]:
+    """
+    Extract a attribute key of a given path and return it normalize with the end
+    of the path.
+
+    Raise an ValueError if no keyword is found.
+    """
+    match_string: Optional[re.Match[str]] = re.search(
+        "^" + LITERAL_STRING_REGEX, path, re.DOTALL
+    )
+    if isinstance(match_string, re.Match):
+        # return the strin attribute
+        return (str(match_string.group()), path[len(match_string.group()) :])
+    match_keyword: Optional[re.Match] = re.search("^" + KEYWORD_REGEX, path)
+    if isinstance(match_keyword, re.Match):
+        # add comma to normalize names
+        return (f'"{match_keyword.group()}"', path[len(match_keyword.group()) :])
+    # only literal string and keywork can be key of set
+    raise ValueError("no attribute key found" + path)
+
+
+def nix_attribute2py_list(attribute: str) -> List[str]:
+    """
+    Extract a nix attribute path into a list path of key.
+    """
+    # attribute don't start with a "."
+    if attribute.startswith("."):
+        raise ValueError("flake attribute can't start with a '.'")
+
+    # list of the path word
+    keys: List[str] = []
+    # the path that will be shorten
+    path: str = attribute
+    # while we have word
+    while path:
+        # get next attribute key
+        key, path = _extract_key(path)
+        # add the new attribute key to the list
+        keys.append(key)
+        # if it's the end quit
+        if not path:
+            break
+        # check that every attribute key is speparated with a '.'
+        if not path.startswith("."):
+            raise ValueError("attribute keys must be separed by a '.': " + attribute)
+        # remove separating point
+        path = path[1:]
+
+    # return Every keys founds
+    return keys

nixops/script_defs.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 
-from nixops.nix_expr import py2nix
+from pathlib import Path
+from urllib.parse import ParseResult, urlparse, unquote
+from nixops.nix_expr import nix_attribute2py_list, py2nix
 from nixops.parallel import run_tasks
 from nixops.storage import StorageBackend, StorageInterface
 from nixops.locks import LockDriver, LockInterface
@@ -37,27 +39,77 @@
 
 
 def get_network_file(args: Namespace) -> NetworkFile:
-    network_dir: str = os.path.abspath(args.network_dir)
+    # Check that we don't try to build flake and classic nix at the same time
+    if args.network_dir != None and args.flake != None:
+        raise ValueError("Both --network and --flake can't be set simultany")
 
+    # We use flake.
+    if args.flake != None:
+        flake: str = args.flake
+        url: ParseResult = urlparse(flake)
+
+        # Get the attribute or default if there is none
+        quote_attribute = url.fragment if url.fragment else "default"
+        # Decode % encoded
+        attribute = unquote(quote_attribute)
+
+        path: str = url.path
+        # If it's a file or a directory get the absolute path
+        if url.scheme in ["file", "path", ""]:
+            # resolve it
+            path = str(Path(path).absolute())
+
+        # Create new url with absolute path and remove fragment (attribute)
+        url = ParseResult(url.scheme, url.netloc, path, url.params, url.query, "")
+
+        # Get the reference without the attribute
+        reference = url.geturl()
+
+        # split the path to pass it to the nix expression
+        attribute_list: List[str] = nix_attribute2py_list(attribute)
+        attribute_path: str = "[ " + " ".join(attribute_list) + " ]"
+
+        # create the network with the reference and the attribute
+        return NetworkFile(reference, attribute_path)
+
+    # we don't use flake.
+
+    # default value of network_dir is None in args but it's current working
+    # dirrectory
+    network_dir_name: str = os.getcwd() if args.network_dir == None else args.network_dir
+    # get real path
+    network_dir: str = os.path.abspath(network_dir_name)
+
+    # check that the folder exist
     if not os.path.exists(network_dir):
         raise ValueError(f"{network_dir} does not exist")
 
+    # path to the classic entry point file
     classic_path = os.path.join(network_dir, "nixops.nix")
+    # path to the flake entry point file
     flake_path = os.path.join(network_dir, "flake.nix")
 
+    # check existing
     classic_exists: bool = os.path.exists(classic_path)
     flake_exists: bool = os.path.exists(flake_path)
 
+    # don't decide for the user, raise an exception.
     if all((flake_exists, classic_exists)):
         raise ValueError("Both flake.nix and nixops.nix cannot coexist")
 
     if classic_exists:
-        return NetworkFile(network=classic_path, is_flake=False)
+        # just return the network with no flake
+        return NetworkFile(network=classic_path, attribute=None)
 
     if flake_exists:
-        return NetworkFile(network=network_dir, is_flake=True)
+        # return the flake path as network and the output attibute.
+        # TODO: depricate this version in favor of the --flake
+        return NetworkFile(network=network_dir, attribute='["default"]')
 
-    raise ValueError(f"Neither flake.nix nor nixops.nix exists in {network_dir}")
+    # it's nether a flake or a classic build.
+    raise ValueError(
+        f"Flake not provided and neither flake.nix nor nixops.nix exists in {network_dir}"
+    )
 
 
 def set_common_depl(depl: nixops.deployment.Deployment, args: Namespace) -> None:
@@ -1168,9 +1220,17 @@ def add_subparser(
         "--network",
         dest="network_dir",
         metavar="FILE",
-        default=os.getcwd(),
+        default=None,
         help="path to a directory containing either nixops.nix or flake.nix",
     )
+    subparser.add_argument(
+        "--flake",
+        "-f",
+        dest="flake",
+        metavar="FLAKE_URI",
+        default=os.environ.get("NIXOPS_FLAKE", None),
+        help="the flake uri.",
+    )
     subparser.add_argument(
         "--deployment",
         "-d",