yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

clan-lib: Move nix_options from Machine class to Flake class

Browse Source
This commit is contained in:
Qubasa
2025-06-23 13:46:10 +02:00
parent 20080f8857
commit cd1d49b603
11 changed files with 57 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
pkgs/clan-cli/clan_cli/__init__.py
View File

@@ -35,20 +35,27 @@ with contextlib.suppress(ImportError):
import argcomplete # type: ignore[no-redef]
def flake_path(arg: str) -> Flake:
def flake_path(arg: str) -> str:
flake_dir = Path(arg).resolve()
if flake_dir.exists() and flake_dir.is_dir():
return Flake(str(flake_dir))
return Flake(arg)
return str(flake_dir)
return arg
def default_flake() -> Flake | None:
def default_flake() -> str | None:
val = get_clan_flake_toplevel_or_env()
if val:
return Flake(str(val))
return str(val)
return None
def create_flake_from_args(args: argparse.Namespace) -> Flake:
"""Create a Flake object from parsed arguments, including nix_options."""
flake_path_str = args.flake
nix_options = getattr(args, "option", [])
return Flake(flake_path_str, nix_options=nix_options)
def add_common_flags(parser: argparse.ArgumentParser) -> None:
def argument_exists(parser: argparse.ArgumentParser, arg: str) -> bool:
"""
@@ -450,6 +457,10 @@ def main() -> None:
if not hasattr(args, "func"):
return
# Convert flake path to Flake object with nix_options if flake argument exists
if hasattr(args, "flake") and args.flake is not None:
args.flake = create_flake_from_args(args)
try:
args.func(args)
except ClanError as e:

2
pkgs/clan-cli/clan_cli/machines/install.py
View File

@@ -42,7 +42,7 @@ def install_command(args: argparse.Namespace) -> None:
else:
password = None
machine = Machine(name=args.machine, flake=args.flake, nix_options=args.option)
machine = Machine(name=args.machine, flake=args.flake)
host_key_check = args.host_key_check
if target_host_str is not None:

4
pkgs/clan-cli/clan_cli/machines/update.py
View File

@@ -46,9 +46,7 @@ def update_command(args: argparse.Namespace) -> None:
raise ClanError(msg)
for machine_name in selected_machines:
machine = Machine(
name=machine_name, flake=args.flake, nix_options=args.option
)
machine = Machine(name=machine_name, flake=args.flake)
machines.append(machine)
if args.target_host is not None and len(machines) > 1:

6
pkgs/clan-cli/clan_cli/tests/helpers/cli.py
View File

@@ -2,7 +2,7 @@ import argparse
import logging
import shlex
from clan_cli import create_parser
from clan_cli import create_flake_from_args, create_parser
from clan_lib.custom_logger import print_trace
log = logging.getLogger(__name__)
@@ -13,6 +13,10 @@ def run(args: list[str]) -> argparse.Namespace:
parsed = parser.parse_args(args)
cmd = shlex.join(["clan", *args])
# Convert flake path to Flake object with nix_options if flake argument exists
if hasattr(parsed, "flake") and parsed.flake is not None:
parsed.flake = create_flake_from_args(parsed)
print_trace(f"$ {cmd}", log, "localhost")
if hasattr(parsed, "func"):
parsed.func(parsed)

2
pkgs/clan-cli/clan_cli/vars/generate.py
View File

@@ -511,7 +511,7 @@ def generate_command(args: argparse.Namespace) -> None:
msg = "Could not find clan flake toplevel directory"
raise ClanError(msg)
machines: list[Machine] = list(list_full_machines(args.flake, args.option).values())
machines: list[Machine] = list(list_full_machines(args.flake).values())
if len(args.machines) > 0:
machines = list(

37
pkgs/clan-cli/clan_lib/flake/flake.py
View File

@@ -576,6 +576,7 @@ class Flake:
identifier: str
hash: str | None = None
store_path: str | None = None
nix_options: list[str] | None = None
_flake_cache_path: Path | None = field(init=False, default=None)
_cache: FlakeCache | None = field(init=False, default=None)
@@ -583,8 +584,13 @@ class Flake:
_is_local: bool | None = field(init=False, default=None)
@classmethod
def from_json(cls: type["Flake"], data: dict[str, Any]) -> "Flake":
return cls(data["identifier"])
def from_json(
cls: type["Flake"],
data: dict[str, Any],
*,
nix_options: list[str] | None = None,
) -> "Flake":
return cls(data["identifier"], nix_options=nix_options)
def __str__(self) -> str:
return self.identifier
@@ -632,10 +638,14 @@ class Flake:
nix_command,
)
if self.nix_options is None:
self.nix_options = []
cmd = [
"flake",
"prefetch",
"--json",
*self.nix_options,
"--option",
"flake-registry",
"",
@@ -690,7 +700,6 @@ class Flake:
def get_from_nix(
self,
selectors: list[str],
nix_options: list[str] | None = None,
apply: str = "v: v",
) -> None:
"""
@@ -722,8 +731,7 @@ class Flake:
self.invalidate_cache()
assert self._cache is not None
if nix_options is None:
nix_options = []
nix_options = self.nix_options if self.nix_options is not None else []
str_selectors: list[str] = []
for selector in selectors:
@@ -736,7 +744,9 @@ class Flake:
# method to getting the NAR hash
fallback_nixpkgs_hash = "@fallback_nixpkgs_hash@"
if not fallback_nixpkgs_hash.startswith("sha256-"):
fallback_nixpkgs = Flake(str(nixpkgs_source()))
fallback_nixpkgs = Flake(
str(nixpkgs_source()), nix_options=self.nix_options
)
fallback_nixpkgs.invalidate_cache()
assert fallback_nixpkgs.hash is not None, (
"this should be impossible as invalidate_cache() should always set `hash`"
@@ -745,7 +755,7 @@ class Flake:
select_hash = "@select_hash@"
if not select_hash.startswith("sha256-"):
select_flake = Flake(str(select_source()))
select_flake = Flake(str(select_source()), nix_options=self.nix_options)
select_flake.invalidate_cache()
assert select_flake.hash is not None, (
"this should be impossible as invalidate_cache() should always set `hash`"
@@ -808,11 +818,7 @@ class Flake:
if self.flake_cache_path:
self._cache.save_to_file(self.flake_cache_path)
def precache(
self,
selectors: list[str],
nix_options: list[str] | None = None,
) -> None:
def precache(self, selectors: list[str]) -> None:
"""
Ensures that the specified selectors are cached locally.
@@ -822,7 +828,6 @@ class Flake:
Args:
selectors (list[str]): A list of attribute selectors to check and cache.
nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
"""
if self._cache is None:
self.invalidate_cache()
@@ -833,12 +838,11 @@ class Flake:
if not self._cache.is_cached(selector):
not_fetched_selectors.append(selector)
if not_fetched_selectors:
self.get_from_nix(not_fetched_selectors, nix_options)
self.get_from_nix(not_fetched_selectors)
def select(
self,
selector: str,
nix_options: list[str] | None = None,
apply: str = "v: v",
) -> Any:
"""
@@ -847,7 +851,6 @@ class Flake:
Args:
selector (str): The attribute selector string to fetch the value for.
nix_options (list[str] | None): Optional additional options to pass to the Nix build command.
"""
if self._cache is None:
self.invalidate_cache()
@@ -856,6 +859,6 @@ class Flake:
if not self._cache.is_cached(selector):
log.debug(f"Cache miss for {selector}")
self.get_from_nix([selector], nix_options, apply=apply)
self.get_from_nix([selector], apply=apply)
value = self._cache.select(selector)
return value

9
pkgs/clan-cli/clan_lib/flake/flake_cache_test.py
View File

@@ -284,13 +284,14 @@ def test_cache_gc(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
}
""")
my_flake = Flake(str(tmp_path / "flake"))
my_flake = Flake(
str(tmp_path / "flake"),
nix_options=["--sandbox-build-dir", str(tmp_path / "build")],
)
if platform == "darwin":
my_flake.select("testfile")
else:
my_flake.select(
"testfile", nix_options=["--sandbox-build-dir", str(tmp_path / "build")]
)
my_flake.select("testfile")
assert my_flake._cache is not None # noqa: SLF001
assert my_flake._cache.is_cached("testfile") # noqa: SLF001
subprocess.run(["nix-collect-garbage"], check=True)

13
pkgs/clan-cli/clan_lib/machines/list.py
View File

@@ -17,9 +17,7 @@ from clan_lib.nix_models.clan import InventoryMachine
log = logging.getLogger(__name__)
def list_full_machines(
flake: Flake, nix_options: list[str] | None = None
) -> dict[str, Machine]:
def list_full_machines(flake: Flake) -> dict[str, Machine]:
"""
Like `list_machines`, but returns a full 'machine' instance for each machine.
"""
@@ -27,9 +25,6 @@ def list_full_machines(
res: dict[str, Machine] = {}
if nix_options is None:
nix_options = []
for inv_machine in machines.values():
name = inv_machine.get("name")
# Technically, this should not happen, but we are defensive here.
@@ -37,11 +32,7 @@ def list_full_machines(
msg = "InternalError: Machine name is required. But got a machine without a name."
raise ClanError(msg)
machine = Machine(
name=name,
flake=flake,
nix_options=nix_options,
)
machine = Machine(name=name, flake=flake)
res[machine.name] = machine
return res

7
pkgs/clan-cli/clan_lib/machines/machines.py
View File

@@ -2,7 +2,7 @@ import importlib
import json
import logging
import re
from dataclasses import dataclass, field
from dataclasses import dataclass
from functools import cached_property
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal
@@ -30,8 +30,6 @@ class Machine:
name: str
flake: Flake
nix_options: list[str] = field(default_factory=list)
def get_inv_machine(self) -> "InventoryMachine":
return get_machine(self.flake, self.name)
@@ -177,8 +175,7 @@ class Machine:
system = config["system"]
return self.flake.select(
f'clanInternals.machines."{system}"."{self.name}".{attr}',
nix_options=nix_options,
f'clanInternals.machines."{system}"."{self.name}".{attr}'
)
def eval_nix(

4
pkgs/clan-cli/clan_lib/machines/update.py
View File

@@ -124,6 +124,8 @@ def deploy_machine(
path = upload_sources(machine, sudo_host)
nix_options = machine.flake.nix_options if machine.flake.nix_options else []
nix_options = [
"--show-trace",
"--option",
@@ -133,7 +135,7 @@ def deploy_machine(
"accept-flake-config",
"true",
"-L",
*machine.nix_options,
*nix_options,
"--flake",
f"{path}#{machine.name}",
]

6
pkgs/clan-cli/clan_lib/persist/inventory_store.py
View File

@@ -86,11 +86,7 @@ class WriteInfo:
class FlakeInterface(Protocol):
def select(
self,
selector: str,
nix_options: list[str] | None = None,
) -> Any: ...
def select(self, selector: str) -> Any: ...
def invalidate_cache(self) -> None: ...