randogoth/curator
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

added env var support

Browse source
This commit is contained in:
randogoth 2026-01-01 20:21:38 +02:00
parent 7d51de1052
commit 25c80fdf3b
3 changed files with 390 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

33
README.md
View file

@ -66,8 +66,9 @@ This command:
2. Disables COPR repositories that are no longer configured
3. Installs all packages listed in `inventory.toml`
4. Creates symlinks for all configured dotfiles
5. Creates backups of existing files before replacing them
6. Updates the last switch timestamp
5. Applies environment variables to `~/.config/environment.d/20-curator.conf` and imports them into the user session
6. Creates backups of existing files before replacing them
7. Updates the last switch timestamp
### `status`
Show current configuration status and information.
@ -117,6 +118,15 @@ Remove the rollback snapshot (`inventory.toml.rollback`).
curator reset
```
### `env`
Apply environment variables defined in `[variables]` and propagate them to systemd/DBus. For your current shell, run:
```bash
curator env
# then, to update this shell:
eval "$(curator env --eval)"
```
## Configuration
### inventory.toml
@ -156,6 +166,10 @@ last_switch = ""
# nixpkgs#git # or just "git" (curator will prefix nixpkgs#)
# nixpkgs#htop # or just "htop"
[variables]
# ENV_VAR = "value"
# ANOTHER = "another value"
[dotfiles]
# Dotfiles to manage with symlinks
# Format: "target_path" = "source_path"
@ -217,6 +231,21 @@ List of rpm-ostree layered packages. **One package per line**—presence means i
#### `[nix]`
List of nix packages. **One package per line**—presence means install. You can specify plain names (e.g., `neovim`) or `nixpkgs#name`; curator will prefix `nixpkgs#` for installs and use the base name for removals.
#### `[variables]`
User environment variables to set via systemd `environment.d`. curator writes them to `~/.config/environment.d/20-curator.conf` during `curator switch`; new sessions will load them automatically.
**Examples:**
```toml
[variables]
EDITOR = "nvim"
PAGER = "less -R"
```
Notes:
- Values are imported into the user systemd and DBus environments during `curator switch` or `curator env`.
- To update a currently running shell, run `eval "$(curator env --eval)"` after applying.
- Later shell init files (e.g., `.bashrc`, `.zshrc`) can still override these values.
#### `[dotfiles]`
Dotfile mappings using symlinks:
- **Key**: Target path where symlink should be created (relative to home directory)

262
src/curator/cli.py
View file

@ -51,6 +51,9 @@ last_switch = ""
# nixpkgs#git
# nixpkgs#htop
[variables]
# EXAMPLE_VAR = "value"
[dotfiles]
# Dotfiles to manage with symlinks
# Format: "target_path" = "source_path"
@ -77,6 +80,7 @@ class CuratorConfig:
flatpak_refs: list[str]
rpm_ostree_packages: list[str]
nix_packages: list[str]
variables: dict[str, str]
dotfiles: dict[str, str]
options: dict[str, Any]
curator_fields: dict[str, Any]
@ -118,6 +122,17 @@ def parse_value(value: str) -> Any:
return strip_quotes(value)
def parse_key_value(entry: str) -> Optional[tuple[str, str]]:
if "=" not in entry:
return None
key_raw, value_raw = entry.split("=", 1)
key = strip_quotes(key_raw.strip())
value = strip_quotes(value_raw.strip())
if not key:
return None
return key, value
def normalize_nix_ref(ref: str) -> str:
ref = ref.strip()
if not ref:
@ -146,6 +161,7 @@ def parse_config(config_path: Path) -> CuratorConfig:
flatpak_refs: list[str] = []
rpm_ostree_packages: list[str] = []
nix_packages: list[str] = []
variables: dict[str, str] = {}
dotfiles: dict[str, str] = {}
options: dict[str, Any] = {}
curator_fields: dict[str, Any] = {}
@ -182,6 +198,16 @@ def parse_config(config_path: Path) -> CuratorConfig:
dnf_packages.append(strip_quotes(entry))
continue
if section == "variables":
if "=" not in line:
continue
key_raw, value_raw = line.split("=", 1)
key = strip_quotes(key_raw.strip())
value = strip_quotes(value_raw.strip())
if key:
variables[key] = value
continue
if section == "dotfiles":
if "=" not in line:
continue
@ -219,6 +245,7 @@ def parse_config(config_path: Path) -> CuratorConfig:
flatpak_refs=flatpak_refs,
rpm_ostree_packages=rpm_ostree_packages,
nix_packages=nix_packages,
variables=variables,
dotfiles=dotfiles,
options=merged_options,
curator_fields=curator_fields,
@ -234,6 +261,7 @@ def empty_config() -> CuratorConfig:
flatpak_refs=[],
rpm_ostree_packages=[],
nix_packages=[],
variables={},
dotfiles={},
options={"backup": True, "backup_dir": "backup"},
curator_fields={},
@ -609,7 +637,7 @@ def gather_nix_packages() -> list[str]:
return sorted(collected)
def backup_target(target_path: Path, backup_dir: Path) -> None:
def backup_target(target_path: Path, backup_dir: Path, *, quiet: bool = False) -> None:
timestamp = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
backup_dir.mkdir(parents=True, exist_ok=True)
backup_path = backup_dir / f"{target_path.name}.{timestamp}.bak"
@ -619,7 +647,8 @@ def backup_target(target_path: Path, backup_dir: Path) -> None:
else:
shutil.copy2(target_path, backup_path)
log_info(f"Backed up {target_path} to {backup_path}")
if not quiet:
log_info(f"Backed up {target_path} to {backup_path}")
def remove_existing(target_path: Path) -> None:
@ -671,6 +700,162 @@ def deploy_dotfiles(config: CuratorConfig, curator_dir: Path) -> None:
log_error(f"Failed to create symlink: {target_path} -> {relative_source} ({exc})")
def _format_env_value(value: str) -> str:
sanitized = value.replace("\r", " ").replace("\n", " ")
escaped = sanitized.replace("\\", "\\\\").replace('"', '\\"')
if re.search(r"\s|#", sanitized) or '"' in sanitized:
return f'"{escaped}"'
return escaped
def _read_environment_file_keys(env_file: Path) -> set[str]:
if not env_file.exists():
return set()
keys: set[str] = set()
try:
for line in env_file.read_text().splitlines():
match = re.match(r"\s*([A-Za-z_][A-Za-z0-9_]*)\s*=", line)
if match:
keys.add(match.group(1))
except OSError:
pass
return keys
def _run_env_command(command: list[str], env: dict[str, str], quiet: bool, description: str) -> None:
try:
result = subprocess.run(command, check=False, env=env, capture_output=True, text=True)
except FileNotFoundError:
if not quiet:
log_warning(f"{description} skipped (command not found): {command[0]}")
return
if result.returncode != 0:
if not quiet:
details = result.stderr.strip() or result.stdout.strip() or "unknown error"
log_warning(f"{description} failed: {details}")
return
if not quiet:
log_success(f"{description} complete")
def propagate_environment_variables(vars_map: dict[str, str], unset_keys: set[str], quiet: bool = False) -> None:
if not vars_map and not unset_keys:
return
env = os.environ.copy()
env.update(vars_map)
set_keys = sorted(vars_map)
unset_list = sorted(unset_keys)
if set_keys:
_run_env_command(
["systemctl", "--user", "import-environment", *set_keys],
env,
quiet,
"Imported environment into systemd --user",
)
_run_env_command(
["dbus-update-activation-environment", "--systemd", *[f"{k}={v}" for k, v in vars_map.items()]],
env,
quiet,
"Updated DBus activation environment",
)
if unset_list:
_run_env_command(
["systemctl", "--user", "unset-environment", *unset_list],
env,
quiet,
"Unset environment in systemd --user",
)
_run_env_command(
["dbus-update-activation-environment", "--systemd", *[f"--unset={key}" for key in unset_list]],
env,
quiet,
"Unset DBus activation environment",
)
def apply_environment_variables(
config: CuratorConfig, curator_dir: Path, *, quiet: bool = False, propagate: bool = False
) -> tuple[dict[str, str], set[str]]:
def log_info_if(message: str) -> None:
if not quiet:
log_info(message)
def log_success_if(message: str) -> None:
if not quiet:
log_success(message)
def log_warning_if(message: str) -> None:
if not quiet:
log_warning(message)
log_info_if("Applying environment variables...")
env_home = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")).expanduser()
env_dir = env_home / "environment.d"
env_file = env_dir / "20-curator.conf"
backup_enabled = bool(config.options.get("backup", True))
backup_dir_name = str(config.options.get("backup_dir", "backup") or "backup")
backup_dir = curator_dir / backup_dir_name
existing_keys = _read_environment_file_keys(env_file)
variables: dict[str, str] = {}
for key, raw_value in config.variables.items():
key_clean = key.strip()
if not key_clean:
log_warning_if("Skipping environment variable with empty name")
continue
variables[key_clean] = str(raw_value)
removed_keys = existing_keys - set(variables.keys())
if not variables:
if env_file.exists():
try:
if backup_enabled:
backup_target(env_file, backup_dir, quiet=quiet)
env_file.unlink()
log_success_if(f"Removed environment file: {env_file}")
except OSError as exc:
log_warning_if(f"Failed to remove environment file {env_file}: {exc}")
else:
log_info_if("No environment variables to apply")
if propagate and (removed_keys or variables):
propagate_environment_variables(variables, removed_keys, quiet)
return variables, removed_keys
lines: list[str] = []
for key, raw_value in sorted(variables.items()):
formatted_value = _format_env_value(str(raw_value))
lines.append(f"{key}={formatted_value}")
env_dir.mkdir(parents=True, exist_ok=True)
try:
if backup_enabled and env_file.exists():
backup_target(env_file, backup_dir, quiet=quiet)
env_file.write_text("\n".join(lines) + "\n")
log_success_if(f"Wrote {len(lines)} environment variables to {env_file}")
except OSError as exc:
if quiet:
print(f"ERROR: Failed to write environment variables to {env_file}: {exc}", file=sys.stderr)
else:
log_error(f"Failed to write environment variables to {env_file}: {exc}")
return variables, removed_keys
if propagate:
propagate_environment_variables(variables, removed_keys, quiet)
return variables, removed_keys
def update_last_switch(config_path: Path, options: dict[str, Any]) -> None:
backup_enabled = bool(options.get("backup", True))
backup_dir_name = str(options.get("backup_dir", "backup") or "backup")
@ -744,6 +929,8 @@ def update_section_entries(config_path: Path, section: str, entries: list[str])
while i < len(lines) and not re.match(r"\s*\[.+]\s*$", lines[i]):
i += 1
output.extend(entries)
if i < len(lines) and (not output or output[-1].strip()):
output.append("")
continue
output.append(line)
i += 1
@ -777,6 +964,8 @@ def merge_section_entries(config_path: Path, section: str, add: set[str], remove
current_entries.update(current_config.copr)
elif section == "dotfiles":
current_entries.update(f"{k} = {v}" for k, v in current_config.dotfiles.items())
elif section == "variables":
current_entries.update(f"{k} = {v}" for k, v in current_config.variables.items())
else:
log_error(f"Unknown section: {section}")
return
@ -785,6 +974,27 @@ def merge_section_entries(config_path: Path, section: str, add: set[str], remove
add_clean = {normalize_nix_ref(item) for item in add_clean}
remove_clean = {normalize_nix_ref(item) for item in remove_clean}
if section == "variables":
current_vars = dict(current_config.variables)
for item in add_clean:
parsed = parse_key_value(item)
if not parsed:
log_warning(f"Skipping invalid variable entry: {item}")
continue
key, value = parsed
current_vars[key] = value
for item in remove_clean:
parsed = parse_key_value(item)
key = parsed[0] if parsed else strip_quotes(item)
key = key.strip()
if key and key in current_vars:
del current_vars[key]
update_section_entries(config_path, section, [f"{k} = {v}" for k, v in sorted(current_vars.items())])
return
new_entries = (current_entries | add_clean) - remove_clean
update_section_entries(config_path, section, sorted(new_entries))
@ -864,6 +1074,7 @@ def _switch(rollback: bool) -> None:
remove_flatpaks(flatpak_removed, flatpak_installed)
remove_rpm_ostree_packages(rpm_ostree_removed, rpm_ostree_installed)
remove_nix_packages(nix_removed_base, set(nix_base_name(p) for p in gather_nix_packages()) if need_nix else None)
apply_environment_variables(current_config, curator_dir, propagate=True)
deploy_dotfiles(current_config, curator_dir)
update_last_switch(curator_toml, current_config.options)
if rollback:
@ -920,9 +1131,12 @@ def apply_inline_updates(entry_args: list[str], add: bool) -> None:
if not section or not value:
log_warning(f"Skipping invalid entry (empty section or value): {arg}")
continue
if section not in {"dnf", "brew", "flatpak", "rpm-ostree", "nix", "copr"}:
if section not in {"dnf", "brew", "flatpak", "rpm-ostree", "nix", "copr", "variables"}:
log_warning(f"Unknown section '{section}' in entry: {arg}")
continue
if section == "variables" and add and "=" not in value:
log_warning(f"Skipping invalid variable entry (expected NAME=VALUE): {arg}")
continue
bucket = changes.setdefault(section, {"add": set(), "remove": set()})
target = "add" if add else "remove"
bucket[target].add(value)
@ -932,6 +1146,31 @@ def apply_inline_updates(entry_args: list[str], add: bool) -> None:
log_success(f"Updated section [{section}] in {curator_toml}")
def env_command(eval_mode: bool = False) -> None:
curator_dir, curator_toml, _, _ = get_paths()
if not curator_toml.exists():
log_error("inventory.toml not found. Run 'curator init' first.")
sys.exit(1)
config = parse_config(curator_toml)
variables, removed_keys = apply_environment_variables(
config, curator_dir, quiet=eval_mode, propagate=True
)
if eval_mode:
exports: list[str] = []
for key, value in sorted(variables.items()):
exports.append(f"export {key}={_format_env_value(value)}")
for key in sorted(removed_keys):
exports.append(f"unset {key}")
if exports:
print("\n".join(exports))
return
log_info("To update your current shell now, run:")
print(' eval "$(curator env --eval)"')
def status_command() -> None:
curator_dir, curator_toml, _, _ = get_paths()
log_info("curator Status")
@ -1006,6 +1245,15 @@ def status_command() -> None:
else:
print(" No nix packages configured")
print()
log_info("Environment Variables:")
if config.variables:
for key, value in sorted(config.variables.items()):
print(f" {key} = {value}")
print(f" Total: {len(config.variables)} environment variables")
else:
print(" No environment variables configured")
print()
log_info("Dotfiles:")
if config.dotfiles:
@ -1040,6 +1288,12 @@ def build_parser() -> argparse.ArgumentParser:
add_parser.add_argument("entries", nargs="+", help="Entries in the form section:value")
remove_parser = subparsers.add_parser("remove", help="Remove entries from inventory.toml (e.g. dnf:uv nix:micro)")
remove_parser.add_argument("entries", nargs="+", help="Entries in the form section:value")
env_parser = subparsers.add_parser("env", help="Apply environment variables immediately")
env_parser.add_argument(
"--eval",
action="store_true",
help='Print export/unset statements for the current shell (use with: eval "$(curator env --eval)")',
)
subparsers.add_parser("status", help="Show current configuration status")
subparsers.add_parser("help", help="Show help message")
return parser
@ -1061,6 +1315,8 @@ def main(argv: list[str] | None = None) -> None:
apply_inline_updates(args.entries, add=True)
elif args.command == "remove":
apply_inline_updates(args.entries, add=False)
elif args.command == "env":
env_command(eval_mode=args.eval)
elif args.command == "status":
status_command()
elif args.command == "help":

103
tests/test_config.py
View file

@ -67,11 +67,14 @@ wget
[flatpak]
org.mozilla.firefox
[rpm-ostree]
podman
[rpm-ostree]
podman
[nix]
nixpkgs#git
[nix]
nixpkgs#git
[variables]
EDITOR = "nvim"
[copr]
copr.fedorainfracloud.org/user/repo
@ -90,6 +93,7 @@ copr.fedorainfracloud.org/user/repo
assert config.nix_packages == ["nixpkgs#git"]
assert config.copr == ["copr.fedorainfracloud.org/user/repo"]
assert config.dotfiles == {".bashrc": ".bashrc"}
assert config.variables == {"EDITOR": "nvim"}
def test_diff_items() -> None:
@ -162,3 +166,94 @@ wget
"copr.fedorainfracloud.org/another/repo",
"copr.fedorainfracloud.org/user/repo",
]
def test_merge_section_entries_preserves_spacing(tmp_path: Path) -> None:
config_path = tmp_path / "inventory.toml"
config_path.write_text(
"""
[dnf]
git
[brew]
wget
""".strip()
)
cli.merge_section_entries(config_path, "dnf", {"vim"}, set())
lines = config_path.read_text().splitlines()
assert "" in lines, "expected to keep a blank line between sections"
dnf_index = lines.index("[dnf]")
brew_index = lines.index("[brew]")
assert lines[dnf_index + 1 : dnf_index + 3] == ["git", "vim"]
assert lines[brew_index - 1] == ""
def test_apply_environment_variables(tmp_path: Path, monkeypatch) -> None:
curator_dir = tmp_path / "curator"
curator_dir.mkdir()
env_home = tmp_path / "xdg"
monkeypatch.setenv("XDG_CONFIG_HOME", str(env_home))
config = cli.empty_config()
config.variables = {"EDITOR": "nvim", "WITH_SPACE": "some value"}
config.options["backup"] = False
written, removed = cli.apply_environment_variables(config, curator_dir)
env_file = env_home / "environment.d" / "20-curator.conf"
assert env_file.exists()
lines = env_file.read_text().splitlines()
assert "EDITOR=nvim" in lines
assert 'WITH_SPACE="some value"' in lines
assert written == {"EDITOR": "nvim", "WITH_SPACE": "some value"}
assert removed == set()
def test_apply_environment_variables_removes_file_when_empty(tmp_path: Path, monkeypatch) -> None:
curator_dir = tmp_path / "curator"
curator_dir.mkdir()
env_home = tmp_path / "xdg"
env_dir = env_home / "environment.d"
env_dir.mkdir(parents=True, exist_ok=True)
env_file = env_dir / "20-curator.conf"
env_file.write_text("OLD=1\n")
monkeypatch.setenv("XDG_CONFIG_HOME", str(env_home))
config = cli.empty_config()
config.options["backup"] = False
variables, removed = cli.apply_environment_variables(config, curator_dir)
assert not env_file.exists()
assert variables == {}
assert removed == {"OLD"}
def test_env_command_eval_outputs_exports(tmp_path: Path, monkeypatch, capsys) -> None:
curator_dir = tmp_path / "curator"
curator_dir.mkdir()
env_file = curator_dir / "inventory.toml"
env_file.write_text(
"""
[variables]
EDITOR = "nvim"
OLD = "1"
""".strip()
)
def fake_get_paths():
return curator_dir, env_file, curator_dir / "last", curator_dir / "rollback"
def fake_apply_environment_variables(config, path, quiet=False, propagate=False):
assert quiet is True
assert propagate is True
return {"EDITOR": "nvim"}, {"OLD"}
monkeypatch.setattr(cli, "get_paths", fake_get_paths)
monkeypatch.setattr(cli, "apply_environment_variables", fake_apply_environment_variables)
cli.env_command(eval_mode=True)
out = capsys.readouterr().out.strip().splitlines()
assert out == ["export EDITOR=nvim", "unset OLD"]