From b7ebc8543c87826009f66c82879055edb9230947 Mon Sep 17 00:00:00 2001 From: Joscha Date: Sat, 30 Aug 2025 23:23:49 +0200 Subject: [PATCH] Add Pacman module --- pasch/__init__.py | 4 +- pasch/cmd.py | 4 +- pasch/modules/__init__.py | 2 + pasch/modules/echo.py | 4 +- pasch/modules/pacman.py | 98 +++++++++++++++++++++++++++++++++++++++ pasch/orchestrator.py | 20 +++++--- 6 files changed, 119 insertions(+), 13 deletions(-) create mode 100644 pasch/modules/pacman.py diff --git a/pasch/__init__.py b/pasch/__init__.py index e6f54ec..57e0299 100644 --- a/pasch/__init__.py +++ b/pasch/__init__.py @@ -1,5 +1,5 @@ from . import modules -from .cmd import run_capture, run_check +from .cmd import run_capture, run_execute from .orchestrator import Module, Orchestrator __all__: list[str] = [ @@ -7,5 +7,5 @@ __all__: list[str] = [ "Orchestrator", "modules", "run_capture", - "run_check", + "run_execute", ] diff --git a/pasch/cmd.py b/pasch/cmd.py index 743d616..8b8decc 100644 --- a/pasch/cmd.py +++ b/pasch/cmd.py @@ -5,12 +5,12 @@ from rich import print from rich.markup import escape -def run_check(*cmd: str) -> None: +def run_execute(*cmd: str) -> None: print(f"[bright_black]$ {escape(shlex.join(cmd))}") subprocess.run(cmd, check=True) def run_capture(*cmd: str) -> str: - print(f"[bright_black]$ {escape(shlex.join(cmd))}") + print(f"[bright_black italic]$ {escape(shlex.join(cmd))}") result = subprocess.run(cmd, check=True, capture_output=True, encoding="utf-8") return result.stdout diff --git a/pasch/modules/__init__.py b/pasch/modules/__init__.py index 6103a92..1ccd902 100644 --- a/pasch/modules/__init__.py +++ b/pasch/modules/__init__.py @@ -1,5 +1,7 @@ from .echo import Echo +from .pacman import Pacman __all__: list[str] = [ "Echo", + "Pacman", ] diff --git a/pasch/modules/echo.py b/pasch/modules/echo.py index 11a3202..bdf1b6b 100644 --- a/pasch/modules/echo.py +++ b/pasch/modules/echo.py @@ -1,4 +1,4 @@ -from pasch.cmd import run_check +from pasch.cmd import run_execute from pasch.orchestrator import Module, Orchestrator @@ -11,4 +11,4 @@ class Echo(Module): self.args.append(arg) def realize(self) -> None: - run_check("echo", *self.args) + run_execute("echo", *self.args) diff --git a/pasch/modules/pacman.py b/pasch/modules/pacman.py new file mode 100644 index 0000000..ab1cebb --- /dev/null +++ b/pasch/modules/pacman.py @@ -0,0 +1,98 @@ +from dataclasses import dataclass, field + +from rich import print +from rich.markup import escape + +from pasch.cmd import run_capture, run_execute +from pasch.orchestrator import Module, Orchestrator + + +@dataclass +class PacmanPackage: + name: str + exclude: set[str] = field(default_factory=set) + + +class Pacman(Module): + def __init__(self, orchestrator: Orchestrator) -> None: + super().__init__(orchestrator) + self.binary: str = "pacman" + self.packages: set[str] = set() + self.excluded: dict[str, set[str]] = {} + + def install(self, *packages: str) -> None: + self.packages.update(packages) + + def exclude(self, group: str, *packages: str) -> None: + self.excluded.setdefault(group, set()).update(packages) + + def realize(self) -> None: + groups = self._get_groups() + + installed = self._get_explicitly_installed_packages() + target = self._resolve_packages(groups, self.packages) + + to_install = target - installed + to_uninstall = installed - target + + for package in sorted(to_install): + print(f"[bold green]+[/] {escape(package)}") + for package in sorted(to_uninstall): + print(f"[bold red]-[/] {escape(package)}") + + self._install_packages(to_install) + self._uninstall_packages(to_uninstall) + + def _pacman_capture(self, *args: str) -> str: + return run_capture(self.binary, *args) + + def _pacman_execute(self, *args: str) -> None: + if self.binary == "paru": + run_execute(self.binary, *args) # Calls sudo itself + else: + run_execute("sudo", self.binary, *args) + + def _get_explicitly_installed_packages(self) -> set[str]: + return set(self._pacman_capture("-Qqe").splitlines()) + + def _get_groups(self) -> dict[str, set[str]]: + groups = {} + for line in self._pacman_capture("-Sgg").splitlines(): + group, package = line.split(" ", maxsplit=1) + groups.setdefault(group, set()).add(package) + return groups + + def _resolve_packages( + self, + groups: dict[str, set[str]], + packages: set[str], + ) -> set[str]: + result = set() + for package in packages: + result.update(self._resolve_package(groups, package)) + return result + + def _resolve_package(self, groups: dict[str, set[str]], package: str) -> set[str]: + packages = groups.get(package) + if packages is None: + return {package} + packages = packages - self.excluded.get(package, set()) + return self._resolve_packages(groups, packages) + + def _install_packages(self, packages: set[str]) -> None: + if self.orchestrator.dry_run: + return + if not packages: + return + + self._pacman_execute("-S", "--needed", *sorted(packages)) + self._pacman_execute("-D", "--asexplicit", *sorted(packages)) + + def _uninstall_packages(self, packages: set[str]) -> None: + if self.orchestrator.dry_run: + return + if not packages: + return + + self._pacman_execute("-D", "--asdeps", *sorted(packages)) + self._pacman_execute("-Rsn", *self._pacman_capture("-Qqdt").splitlines()) diff --git a/pasch/orchestrator.py b/pasch/orchestrator.py index d205629..1a417e0 100644 --- a/pasch/orchestrator.py +++ b/pasch/orchestrator.py @@ -2,6 +2,9 @@ from __future__ import annotations from abc import ABC, abstractmethod +from rich import print +from rich.markup import escape + class Module(ABC): def __init__(self, orchestrator: Orchestrator) -> None: @@ -13,16 +16,19 @@ class Module(ABC): class Orchestrator: - def __init__(self) -> None: - self.frozen: bool = False - self.modules: list[Module] = [] + def __init__(self, dry_run: bool = False) -> None: + self.dry_run = dry_run + + self._frozen: bool = False + self._modules: list[Module] = [] def register(self, module: Module) -> None: - if self.frozen: + if self._frozen: raise Exception("registering module wile orchestrator is frozen") - self.modules.append(module) + self._modules.append(module) def realize(self) -> None: - self.frozen = True - for module in reversed(self.modules): + self._frozen = True + for module in reversed(self._modules): + print(f"[bold bright_magenta]\\[{escape(type(module).__name__)}]") module.realize()