Refactor pkgmgr into modular installer pipeline with Nix flake support, PKGBUILD build workflow, local Nix cache, and full test suite restructuring.

See conversation: https://chatgpt.com/share/69332519-7ff4-800f-bc21-7fcd24a66c10
This commit is contained in:
Kevin Veen-Birkenbach
2025-12-05 19:32:42 +01:00
parent 341ec1179e
commit f5475d86e2
35 changed files with 1684 additions and 524 deletions

30
pkgmgr/context.py Normal file
View File

@@ -0,0 +1,30 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Shared context object for repository installation steps.
This data class bundles all information needed by installer components so
they do not depend on global state or long parameter lists.
"""
from dataclasses import dataclass
from typing import Any, Dict, List
@dataclass
class RepoContext:
"""Container for all repository-related data used during installation."""
repo: Dict[str, Any]
identifier: str
repo_dir: str
repositories_base_dir: str
bin_dir: str
all_repos: List[Dict[str, Any]]
no_verification: bool
preview: bool
quiet: bool
clone_mode: str
update_dependencies: bool

View File

@@ -1,243 +1,203 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Repository installation pipeline for pkgmgr.
This module orchestrates the installation of repositories by:
1. Ensuring the repository directory exists (cloning if necessary).
2. Verifying the repository according to the configured policies.
3. Creating executable links using create_ink().
4. Running a sequence of modular installer components that handle
specific technologies or manifests (pkgmgr.yml, PKGBUILD, Nix,
Ansible requirements, Python, Makefile).
The goal is to keep this file thin and delegate most logic to small,
focused installer classes.
"""
import os
import subprocess
import sys
import tempfile
import shutil
import yaml
from typing import List, Dict, Any, Tuple
from pkgmgr.get_repo_identifier import get_repo_identifier
from pkgmgr.get_repo_dir import get_repo_dir
from pkgmgr.create_ink import create_ink
from pkgmgr.run_command import run_command
from pkgmgr.verify import verify_repository
from pkgmgr.clone_repos import clone_repos
from pkgmgr.context import RepoContext
def _extract_pkgbuild_array(repo_dir: str, var_name: str) -> list:
"""
Extract a Bash array (depends/makedepends) from PKGBUILD using bash itself.
Returns a list of package names or an empty list on error.
"""
pkgbuild_path = os.path.join(repo_dir, "PKGBUILD")
if not os.path.exists(pkgbuild_path):
return []
# Installer implementations
from pkgmgr.installers.pkgmgr_manifest import PkgmgrManifestInstaller
from pkgmgr.installers.pkgbuild import PkgbuildInstaller
from pkgmgr.installers.nix_flake import NixFlakeInstaller
from pkgmgr.installers.ansible_requirements import AnsibleRequirementsInstaller
from pkgmgr.installers.python import PythonInstaller
from pkgmgr.installers.makefile import MakefileInstaller
from pkgmgr.installers.aur import AurInstaller
script = f'source PKGBUILD >/dev/null 2>&1; printf "%s\\n" "${{{var_name}[@]}}"'
try:
output = subprocess.check_output(
["bash", "-lc", script],
cwd=repo_dir,
text=True,
# Ordered list of installers to apply to each repository
INSTALLERS = [
PkgmgrManifestInstaller(),
PkgbuildInstaller(),
NixFlakeInstaller(),
AnsibleRequirementsInstaller(),
PythonInstaller(),
MakefileInstaller(),
AurInstaller(),
]
def _ensure_repo_dir(
repo: Dict[str, Any],
repositories_base_dir: str,
all_repos: List[Dict[str, Any]],
preview: bool,
no_verification: bool,
clone_mode: str,
identifier: str,
) -> str:
"""
Ensure the repository directory exists. If not, attempt to clone it.
Returns the repository directory path or an empty string if cloning failed.
"""
repo_dir = get_repo_dir(repositories_base_dir, repo)
if not os.path.exists(repo_dir):
print(f"Repository directory '{repo_dir}' does not exist. Cloning it now...")
clone_repos(
[repo],
repositories_base_dir,
all_repos,
preview,
no_verification,
clone_mode,
)
except Exception:
return []
if not os.path.exists(repo_dir):
print(f"Cloning failed for repository {identifier}. Skipping installation.")
return ""
return [line.strip() for line in output.splitlines() if line.strip()]
return repo_dir
def _install_arch_dependencies_from_pkgbuild(repo_dir: str, preview: bool) -> None:
"""
If PKGBUILD exists and pacman is available, install depends + makedepends
via pacman.
"""
if shutil.which("pacman") is None:
return
pkgbuild_path = os.path.join(repo_dir, "PKGBUILD")
if not os.path.exists(pkgbuild_path):
return
depends = _extract_pkgbuild_array(repo_dir, "depends")
makedepends = _extract_pkgbuild_array(repo_dir, "makedepends")
all_pkgs = depends + makedepends
if not all_pkgs:
return
cmd = "sudo pacman -S --noconfirm " + " ".join(all_pkgs)
run_command(cmd, preview=preview)
def _install_nix_flake_profile(repo_dir: str, preview: bool) -> None:
"""
If flake.nix exists and 'nix' is available, try to install a profile
from the flake. Convention: try .#pkgmgr, then .#default.
"""
flake_path = os.path.join(repo_dir, "flake.nix")
if not os.path.exists(flake_path):
return
if shutil.which("nix") is None:
print("Warning: flake.nix found but 'nix' command not available. Skipping flake setup.")
return
print("Nix flake detected, attempting to install profile output...")
for output in ("pkgmgr", "default"):
cmd = f"nix profile install {repo_dir}#{output}"
try:
run_command(cmd, preview=preview)
print(f"Nix flake output '{output}' successfully installed.")
break
except SystemExit as e:
print(f"[Warning] Failed to install Nix flake output '{output}': {e}")
def _install_pkgmgr_dependencies_from_manifest(
def _verify_repo(
repo: Dict[str, Any],
repo_dir: str,
no_verification: bool,
update_dependencies: bool,
clone_mode: str,
identifier: str,
) -> bool:
"""
Verify the repository using verify_repository().
Returns True if installation should proceed, False if it should be skipped.
"""
verified_info = repo.get("verified")
verified_ok, errors, commit_hash, signing_key = verify_repository(
repo,
repo_dir,
mode="local",
no_verification=no_verification,
)
if not no_verification and verified_info and not verified_ok:
print(f"Warning: Verification failed for {identifier}:")
for err in errors:
print(f" - {err}")
choice = input("Proceed with installation? (y/N): ").strip().lower()
if choice != "y":
print(f"Skipping installation for {identifier}.")
return False
return True
def _create_context(
repo: Dict[str, Any],
identifier: str,
repo_dir: str,
repositories_base_dir: str,
bin_dir: str,
all_repos: List[Dict[str, Any]],
no_verification: bool,
preview: bool,
) -> None:
quiet: bool,
clone_mode: str,
update_dependencies: bool,
) -> RepoContext:
"""
Read pkgmgr.yml (if present) and install referenced pkgmgr repository
dependencies.
Expected format:
version: 1
author: "..."
url: "..."
description: "..."
dependencies:
- repository: github:user/repo
version: main
reason: "Optional description"
Build a RepoContext for the given repository and parameters.
"""
manifest_path = os.path.join(repo_dir, "pkgmgr.yml")
if not os.path.exists(manifest_path):
return
try:
with open(manifest_path, "r", encoding="utf-8") as f:
manifest = yaml.safe_load(f) or {}
except Exception as e:
print(f"Error loading pkgmgr.yml in '{repo_dir}': {e}")
return
dependencies = manifest.get("dependencies", []) or []
if not isinstance(dependencies, list) or not dependencies:
return
# Optional: show basic metadata (author/url/description) if present
author = manifest.get("author")
url = manifest.get("url")
description = manifest.get("description")
if not preview:
print("pkgmgr manifest detected:")
if author:
print(f" author: {author}")
if url:
print(f" url: {url}")
if description:
print(f" description: {description}")
dep_repo_ids = []
for dep in dependencies:
if not isinstance(dep, dict):
continue
repo_id = dep.get("repository")
if repo_id:
dep_repo_ids.append(str(repo_id))
# Optionally: update (pull) dependencies before installing
if update_dependencies and dep_repo_ids:
cmd_pull = "pkgmgr pull " + " ".join(dep_repo_ids)
try:
run_command(cmd_pull, preview=preview)
except SystemExit as e:
print(f"Warning: 'pkgmgr pull' for dependencies failed (exit code {e}).")
# Install dependencies one by one
for dep in dependencies:
if not isinstance(dep, dict):
continue
repo_id = dep.get("repository")
if not repo_id:
continue
version = dep.get("version")
reason = dep.get("reason")
if reason and not preview:
print(f"Installing dependency {repo_id}: {reason}")
else:
print(f"Installing dependency {repo_id}...")
cmd = f"pkgmgr install {repo_id}"
if version:
cmd += f" --version {version}"
if no_verification:
cmd += " --no-verification"
if update_dependencies:
cmd += " --dependencies"
if clone_mode:
cmd += f" --clone-mode {clone_mode}"
try:
run_command(cmd, preview=preview)
except SystemExit as e:
print(f"[Warning] Failed to install dependency '{repo_id}': {e}")
return RepoContext(
repo=repo,
identifier=identifier,
repo_dir=repo_dir,
repositories_base_dir=repositories_base_dir,
bin_dir=bin_dir,
all_repos=all_repos,
no_verification=no_verification,
preview=preview,
quiet=quiet,
clone_mode=clone_mode,
update_dependencies=update_dependencies,
)
def install_repos(
selected_repos,
repositories_base_dir,
bin_dir,
all_repos,
no_verification,
preview,
quiet,
selected_repos: List[Dict[str, Any]],
repositories_base_dir: str,
bin_dir: str,
all_repos: List[Dict[str, Any]],
no_verification: bool,
preview: bool,
quiet: bool,
clone_mode: str,
update_dependencies: bool,
):
) -> None:
"""
Install repositories by creating symbolic links and processing standard
manifest files (pkgmgr.yml, PKGBUILD, flake.nix, Ansible requirements,
Python manifests, Makefile).
Python manifests, Makefile) via dedicated installer components.
"""
for repo in selected_repos:
repo_identifier = get_repo_identifier(repo, all_repos)
repo_dir = get_repo_dir(repositories_base_dir, repo)
if not os.path.exists(repo_dir):
print(f"Repository directory '{repo_dir}' does not exist. Cloning it now...")
# Pass the clone_mode parameter to clone_repos
clone_repos(
[repo],
repositories_base_dir,
all_repos,
preview,
no_verification,
clone_mode,
)
if not os.path.exists(repo_dir):
print(f"Cloning failed for repository {repo_identifier}. Skipping installation.")
continue
verified_info = repo.get("verified")
verified_ok, errors, commit_hash, signing_key = verify_repository(
repo,
repo_dir,
mode="local",
identifier = get_repo_identifier(repo, all_repos)
repo_dir = _ensure_repo_dir(
repo=repo,
repositories_base_dir=repositories_base_dir,
all_repos=all_repos,
preview=preview,
no_verification=no_verification,
clone_mode=clone_mode,
identifier=identifier,
)
if not repo_dir:
continue
if not _verify_repo(
repo=repo,
repo_dir=repo_dir,
no_verification=no_verification,
identifier=identifier,
):
continue
ctx = _create_context(
repo=repo,
identifier=identifier,
repo_dir=repo_dir,
repositories_base_dir=repositories_base_dir,
bin_dir=bin_dir,
all_repos=all_repos,
no_verification=no_verification,
preview=preview,
quiet=quiet,
clone_mode=clone_mode,
update_dependencies=update_dependencies,
)
if not no_verification and verified_info and not verified_ok:
print(f"Warning: Verification failed for {repo_identifier}:")
for err in errors:
print(f" - {err}")
choice = input("Proceed with installation? (y/N): ").strip().lower()
if choice != "y":
print(f"Skipping installation for {repo_identifier}.")
continue
# Create the symlink using create_ink.
# Create the symlink using create_ink before running installers.
create_ink(
repo,
repositories_base_dir,
@@ -247,77 +207,7 @@ def install_repos(
preview=preview,
)
# 1) pkgmgr.yml (pkgmgr-internal manifest for other repositories)
_install_pkgmgr_dependencies_from_manifest(
repo_dir=repo_dir,
no_verification=no_verification,
update_dependencies=update_dependencies,
clone_mode=clone_mode,
preview=preview,
)
# 2) Arch: PKGBUILD (depends/makedepends)
_install_arch_dependencies_from_pkgbuild(repo_dir, preview=preview)
# 3) Nix: flake.nix
_install_nix_flake_profile(repo_dir, preview=preview)
# 4) Ansible: requirements.yml (only collections/roles)
req_file = os.path.join(repo_dir, "requirements.yml")
if os.path.exists(req_file):
try:
with open(req_file, "r", encoding="utf-8") as f:
requirements = yaml.safe_load(f) or {}
except Exception as e:
print(f"Error loading requirements.yml in {repo_identifier}: {e}")
requirements = None
if requirements and isinstance(requirements, dict):
if "collections" in requirements or "roles" in requirements:
print(f"Ansible dependencies found in {repo_identifier}, installing...")
ansible_requirements = {}
if "collections" in requirements:
ansible_requirements["collections"] = requirements["collections"]
if "roles" in requirements:
ansible_requirements["roles"] = requirements["roles"]
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".yml",
delete=False,
) as tmp:
yaml.dump(ansible_requirements, tmp, default_flow_style=False)
tmp_filename = tmp.name
if "collections" in ansible_requirements:
print(f"Ansible collections found in {repo_identifier}, installing...")
cmd = f"ansible-galaxy collection install -r {tmp_filename}"
run_command(cmd, cwd=repo_dir, preview=preview)
if "roles" in ansible_requirements:
print(f"Ansible roles found in {repo_identifier}, installing...")
cmd = f"ansible-galaxy role install -r {tmp_filename}"
run_command(cmd, cwd=repo_dir, preview=preview)
# 5) Python: pyproject.toml (modern) / requirements.txt (classic)
pyproject_path = os.path.join(repo_dir, "pyproject.toml")
if os.path.exists(pyproject_path):
print(f"pyproject.toml found in {repo_identifier}, installing Python project...")
cmd = "~/.venvs/pkgmgr/bin/pip install ."
run_command(cmd, cwd=repo_dir, preview=preview)
req_txt_file = os.path.join(repo_dir, "requirements.txt")
if os.path.exists(req_txt_file):
print(f"requirements.txt found in {repo_identifier}, installing Python dependencies...")
cmd = "~/.venvs/pkgmgr/bin/pip install -r requirements.txt"
run_command(cmd, cwd=repo_dir, preview=preview)
# 6) Makefile: make install (if present)
makefile_path = os.path.join(repo_dir, "Makefile")
if os.path.exists(makefile_path):
cmd = "make install"
try:
run_command(cmd, cwd=repo_dir, preview=preview)
except SystemExit as e:
print(f"[Warning] Failed to run '{cmd}' for {repo_identifier}: {e}")
# Run all installers that support this repository.
for installer in INSTALLERS:
if installer.supports(ctx):
installer.run(ctx)

View File

@@ -0,0 +1,9 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer package for pkgmgr.
Each installer implements a small, focused step in the repository
installation pipeline (e.g. PKGBUILD dependencies, Nix flakes, Python, etc.).
"""

View File

@@ -0,0 +1,71 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer for Ansible dependencies defined in requirements.yml.
This installer installs collections and roles via ansible-galaxy when found.
"""
import os
import tempfile
from typing import Any, Dict
import yaml
from pkgmgr.context import RepoContext
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.run_command import run_command
class AnsibleRequirementsInstaller(BaseInstaller):
"""Install Ansible collections and roles from requirements.yml."""
REQUIREMENTS_FILE = "requirements.yml"
def supports(self, ctx: RepoContext) -> bool:
req_file = os.path.join(ctx.repo_dir, self.REQUIREMENTS_FILE)
return os.path.exists(req_file)
def _load_requirements(self, req_path: str, identifier: str) -> Dict[str, Any]:
try:
with open(req_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f) or {}
except Exception as exc:
print(f"Error loading {self.REQUIREMENTS_FILE} in {identifier}: {exc}")
return {}
def run(self, ctx: RepoContext) -> None:
req_file = os.path.join(ctx.repo_dir, self.REQUIREMENTS_FILE)
requirements = self._load_requirements(req_file, ctx.identifier)
if not requirements or not isinstance(requirements, dict):
return
if "collections" not in requirements and "roles" not in requirements:
return
print(f"Ansible dependencies found in {ctx.identifier}, installing...")
ansible_requirements: Dict[str, Any] = {}
if "collections" in requirements:
ansible_requirements["collections"] = requirements["collections"]
if "roles" in requirements:
ansible_requirements["roles"] = requirements["roles"]
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".yml",
delete=False,
) as tmp:
yaml.dump(ansible_requirements, tmp, default_flow_style=False)
tmp_filename = tmp.name
if "collections" in ansible_requirements:
print(f"Ansible collections found in {ctx.identifier}, installing...")
cmd = f"ansible-galaxy collection install -r {tmp_filename}"
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
if "roles" in ansible_requirements:
print(f"Ansible roles found in {ctx.identifier}, installing...")
cmd = f"ansible-galaxy role install -r {tmp_filename}"
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)

131
pkgmgr/installers/aur.py Normal file
View File

@@ -0,0 +1,131 @@
# pkgmgr/installers/aur.py
import os
import shutil
import yaml
from typing import List
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.context import RepoContext
from pkgmgr.run_command import run_command
AUR_CONFIG_FILENAME = "aur.yml"
class AurInstaller(BaseInstaller):
"""
Installer for Arch AUR dependencies declared in an `aur.yml` file.
This installer is:
- Arch-only (requires `pacman`)
- optional helper-driven (yay/paru/..)
- safe to ignore on non-Arch systems
"""
def _is_arch_like(self) -> bool:
return shutil.which("pacman") is not None
def _config_path(self, ctx: RepoContext) -> str:
return os.path.join(ctx.repo_dir, AUR_CONFIG_FILENAME)
def _load_config(self, ctx: RepoContext) -> dict:
path = self._config_path(ctx)
if not os.path.exists(path):
return {}
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml.safe_load(f) or {}
except Exception as exc:
print(f"[Warning] Failed to load AUR config from '{path}': {exc}")
return {}
if not isinstance(data, dict):
print(f"[Warning] AUR config '{path}' is not a mapping. Ignoring.")
return {}
return data
def _get_helper(self, cfg: dict) -> str:
# Priority: config.helper > $AUR_HELPER > "yay"
helper = cfg.get("helper")
if isinstance(helper, str) and helper.strip():
return helper.strip()
env_helper = os.environ.get("AUR_HELPER")
if env_helper:
return env_helper.strip()
return "yay"
def _get_packages(self, cfg: dict) -> List[str]:
raw = cfg.get("packages", [])
if not isinstance(raw, list):
return []
names: List[str] = []
for entry in raw:
if isinstance(entry, str):
name = entry.strip()
if name:
names.append(name)
elif isinstance(entry, dict):
name = str(entry.get("name", "")).strip()
if name:
names.append(name)
return names
# --- BaseInstaller API -------------------------------------------------
def supports(self, ctx: RepoContext) -> bool:
"""
This installer is supported if:
- We are on an Arch-like system (pacman available),
- An aur.yml exists,
- That aur.yml declares at least one package.
"""
if not self._is_arch_like():
return False
cfg = self._load_config(ctx)
if not cfg:
return False
packages = self._get_packages(cfg)
return len(packages) > 0
def run(self, ctx: RepoContext) -> None:
"""
Install AUR packages using the configured helper (default: yay).
"""
if not self._is_arch_like():
print("AUR installer skipped: not an Arch-like system.")
return
cfg = self._load_config(ctx)
if not cfg:
print("AUR installer: no valid aur.yml found; skipping.")
return
packages = self._get_packages(cfg)
if not packages:
print("AUR installer: no AUR packages defined; skipping.")
return
helper = self._get_helper(cfg)
if shutil.which(helper) is None:
print(
f"[Warning] AUR helper '{helper}' is not available on PATH. "
f"Please install it (e.g. via your aur_builder setup). "
f"Skipping AUR installation."
)
return
pkg_list_str = " ".join(packages)
print(f"Installing AUR packages via '{helper}': {pkg_list_str}")
cmd = f"{helper} -S --noconfirm {pkg_list_str}"
# We respect preview mode to allow dry runs.
run_command(cmd, preview=ctx.preview)

34
pkgmgr/installers/base.py Normal file
View File

@@ -0,0 +1,34 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Base interface for all installer components in the pkgmgr installation pipeline.
"""
from abc import ABC, abstractmethod
from pkgmgr.context import RepoContext
class BaseInstaller(ABC):
"""
A single step in the installation pipeline for a repository.
Implementations should be small and focused on one technology or manifest
type (e.g. PKGBUILD, Nix, Python, Ansible).
"""
@abstractmethod
def supports(self, ctx: RepoContext) -> bool:
"""
Return True if this installer should run for the given repository
context. This is typically based on file existence or platform checks.
"""
raise NotImplementedError
@abstractmethod
def run(self, ctx: RepoContext) -> None:
"""
Execute the installer logic for the given repository context.
Implementations may raise SystemExit via run_command() on errors.
"""
raise NotImplementedError

View File

@@ -0,0 +1,32 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer that triggers `make install` if a Makefile is present.
This is useful for repositories that expose a standard Makefile-based
installation step.
"""
import os
from pkgmgr.context import RepoContext
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.run_command import run_command
class MakefileInstaller(BaseInstaller):
"""Run `make install` if a Makefile exists in the repository."""
MAKEFILE_NAME = "Makefile"
def supports(self, ctx: RepoContext) -> bool:
makefile_path = os.path.join(ctx.repo_dir, self.MAKEFILE_NAME)
return os.path.exists(makefile_path)
def run(self, ctx: RepoContext) -> None:
cmd = "make install"
try:
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
except SystemExit as exc:
print(f"[Warning] Failed to run '{cmd}' for {ctx.identifier}: {exc}")

View File

@@ -0,0 +1,47 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer for Nix flakes.
If a repository contains flake.nix and the 'nix' command is available, this
installer will try to install a profile output from the flake.
"""
import os
import shutil
from pkgmgr.context import RepoContext
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.run_command import run_command
class NixFlakeInstaller(BaseInstaller):
"""Install Nix flake profiles for repositories that define flake.nix."""
FLAKE_FILE = "flake.nix"
def supports(self, ctx: RepoContext) -> bool:
if shutil.which("nix") is None:
return False
flake_path = os.path.join(ctx.repo_dir, self.FLAKE_FILE)
return os.path.exists(flake_path)
def run(self, ctx: RepoContext) -> None:
flake_path = os.path.join(ctx.repo_dir, self.FLAKE_FILE)
if not os.path.exists(flake_path):
return
if shutil.which("nix") is None:
print("Warning: flake.nix found but 'nix' command not available. Skipping flake setup.")
return
print("Nix flake detected, attempting to install profile output...")
for output in ("pkgmgr", "default"):
cmd = f"nix profile install {ctx.repo_dir}#{output}"
try:
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
print(f"Nix flake output '{output}' successfully installed.")
except SystemExit as e:
print(f"[Warning] Failed to install Nix flake output '{output}': {e}")

View File

@@ -0,0 +1,71 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer for Arch Linux dependencies defined in PKGBUILD files.
This installer extracts depends/makedepends from PKGBUILD and installs them
via pacman on Arch-based systems.
"""
import os
import shutil
import subprocess
from typing import List
from pkgmgr.context import RepoContext
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.run_command import run_command
class PkgbuildInstaller(BaseInstaller):
"""Install Arch dependencies (depends/makedepends) from PKGBUILD."""
PKGBUILD_NAME = "PKGBUILD"
def supports(self, ctx: RepoContext) -> bool:
if shutil.which("pacman") is None:
return False
pkgbuild_path = os.path.join(ctx.repo_dir, self.PKGBUILD_NAME)
return os.path.exists(pkgbuild_path)
def _extract_pkgbuild_array(self, ctx: RepoContext, var_name: str) -> List[str]:
"""
Extract a Bash array (depends/makedepends) from PKGBUILD using bash itself.
Returns a list of package names or an empty list on error.
Uses a minimal shell environment (no profile/rc) to avoid noise from MOTD
or interactive shell banners polluting the output.
"""
pkgbuild_path = os.path.join(ctx.repo_dir, self.PKGBUILD_NAME)
if not os.path.exists(pkgbuild_path):
return []
script = f'source {self.PKGBUILD_NAME} >/dev/null 2>&1; printf "%s\\n" "${{{var_name}[@]}}"'
try:
output = subprocess.check_output(
["bash", "--noprofile", "--norc", "-c", script],
cwd=ctx.repo_dir,
text=True,
)
except Exception:
return []
packages: List[str] = []
for line in output.splitlines():
line = line.strip()
if not line:
continue
packages.append(line)
return packages
def run(self, ctx: RepoContext) -> None:
depends = self._extract_pkgbuild_array(ctx, "depends")
makedepends = self._extract_pkgbuild_array(ctx, "makedepends")
all_pkgs = depends + makedepends
if not all_pkgs:
return
cmd = "sudo pacman -S --noconfirm " + " ".join(all_pkgs)
run_command(cmd, preview=ctx.preview)

View File

@@ -0,0 +1,114 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Installer for pkgmgr.yml manifest dependencies.
This installer reads pkgmgr.yml (if present) and installs referenced pkgmgr
repository dependencies via pkgmgr itself.
"""
import os
from typing import Any, Dict, List
import yaml
from pkgmgr.context import RepoContext
from pkgmgr.installers.base import BaseInstaller
from pkgmgr.run_command import run_command
class PkgmgrManifestInstaller(BaseInstaller):
"""Install pkgmgr-defined repository dependencies from pkgmgr.yml."""
MANIFEST_NAME = "pkgmgr.yml"
def supports(self, ctx: RepoContext) -> bool:
manifest_path = os.path.join(ctx.repo_dir, self.MANIFEST_NAME)
return os.path.exists(manifest_path)
def _load_manifest(self, manifest_path: str) -> Dict[str, Any]:
try:
with open(manifest_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f) or {}
except Exception as exc:
print(f"Error loading {self.MANIFEST_NAME} in '{manifest_path}': {exc}")
return {}
def _collect_dependency_ids(self, dependencies: List[Dict[str, Any]]) -> List[str]:
ids: List[str] = []
for dep in dependencies:
if not isinstance(dep, dict):
continue
repo_id = dep.get("repository")
if repo_id:
ids.append(str(repo_id))
return ids
def run(self, ctx: RepoContext) -> None:
manifest_path = os.path.join(ctx.repo_dir, self.MANIFEST_NAME)
manifest = self._load_manifest(manifest_path)
if not manifest:
return
dependencies = manifest.get("dependencies", []) or []
if not isinstance(dependencies, list) or not dependencies:
return
author = manifest.get("author")
url = manifest.get("url")
description = manifest.get("description")
if not ctx.preview:
print("pkgmgr manifest detected:")
if author:
print(f" author: {author}")
if url:
print(f" url: {url}")
if description:
print(f" description: {description}")
dep_repo_ids = self._collect_dependency_ids(dependencies)
if ctx.update_dependencies and dep_repo_ids:
cmd_pull = "pkgmgr pull " + " ".join(dep_repo_ids)
try:
run_command(cmd_pull, preview=ctx.preview)
except SystemExit as exc:
print(f"Warning: 'pkgmgr pull' for dependencies failed (exit code {exc}).")
# Install dependencies one by one
for dep in dependencies:
if not isinstance(dep, dict):
continue
repo_id = dep.get("repository")
if not repo_id:
continue
version = dep.get("version")
reason = dep.get("reason")
if reason and not ctx.preview:
print(f"Installing dependency {repo_id}: {reason}")
else:
print(f"Installing dependency {repo_id}...")
cmd = f"pkgmgr install {repo_id}"
if version:
cmd += f" --version {version}"
if ctx.no_verification:
cmd += " --no-verification"
if ctx.update_dependencies:
cmd += " --dependencies"
if ctx.clone_mode:
cmd += f" --clone-mode {ctx.clone_mode}"
try:
run_command(cmd, preview=ctx.preview)
except SystemExit as exc:
print(f"[Warning] Failed to install dependency '{repo_id}': {exc}")

View File

@@ -0,0 +1,89 @@
import os
import sys
from .base import BaseInstaller
from pkgmgr.run_command import run_command
class PythonInstaller(BaseInstaller):
"""
Install Python projects based on pyproject.toml and/or requirements.txt.
Strategy:
- Determine a pip command in this order:
1. $PKGMGR_PIP (explicit override, e.g. ~/.venvs/pkgmgr/bin/pip)
2. sys.executable -m pip (current interpreter)
3. "pip" from PATH as last resort
- If pyproject.toml exists: pip install .
- If requirements.txt exists: pip install -r requirements.txt
"""
name = "python"
def supports(self, ctx) -> bool:
"""
Return True if this installer should handle the given repository.
ctx must provide:
- repo_dir: filesystem path to the repository
"""
repo_dir = ctx.repo_dir
return (
os.path.exists(os.path.join(repo_dir, "pyproject.toml"))
or os.path.exists(os.path.join(repo_dir, "requirements.txt"))
)
def _pip_cmd(self) -> str:
"""
Resolve the pip command to use.
"""
# 1) Explicit override via environment variable
explicit = os.environ.get("PKGMGR_PIP", "").strip()
if explicit:
return explicit
# 2) Current Python interpreter (works well in Nix/dev shells)
if sys.executable:
return f"{sys.executable} -m pip"
# 3) Fallback to plain pip
return "pip"
def run(self, ctx) -> None:
"""
ctx must provide:
- repo_dir: path to repository
- identifier: human readable name
- preview: bool
"""
pip_cmd = self._pip_cmd()
pyproject = os.path.join(ctx.repo_dir, "pyproject.toml")
if os.path.exists(pyproject):
print(
f"pyproject.toml found in {ctx.identifier}, "
f"installing Python project..."
)
cmd = f"{pip_cmd} install ."
try:
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
except SystemExit as exc:
print(
f"[Warning] Failed to install Python project in {ctx.identifier}: {exc}"
)
req_txt = os.path.join(ctx.repo_dir, "requirements.txt")
if os.path.exists(req_txt):
print(
f"requirements.txt found in {ctx.identifier}, "
f"installing Python dependencies..."
)
cmd = f"{pip_cmd} install -r requirements.txt"
try:
run_command(cmd, cwd=ctx.repo_dir, preview=ctx.preview)
except SystemExit as exc:
print(
f"[Warning] Failed to install Python dependencies in {ctx.identifier}: {exc}"
)