From c036ec03cccfe480b3643d1d52ad833c257138ee Mon Sep 17 00:00:00 2001 From: Kivi Kaitaniemi Date: Sun, 14 Dec 2025 23:43:09 +0200 Subject: [PATCH] Add flatpak as a plugin --- src/decman/__init__.py | 8 +- src/decman/app.py | 2 +- src/decman/core/command.py | 2 +- src/decman/plugins/aur/__init__.py | 8 +- src/decman/plugins/flatpak.py | 258 +++++++++++++++++++++++++++++ 5 files changed, 271 insertions(+), 7 deletions(-) create mode 100644 src/decman/plugins/flatpak.py diff --git a/src/decman/__init__.py b/src/decman/__init__.py index 0874660..ccc1385 100644 --- a/src/decman/__init__.py +++ b/src/decman/__init__.py @@ -9,9 +9,10 @@ from decman.core.error import SourceError from decman.core.fs import Directory, File from decman.core.module import Module from decman.plugins import Plugin, available_plugins -from decman.plugins.aur import AUR # Plugin types +from decman.plugins.aur import AUR +from decman.plugins.flatpak import Flatpak from decman.plugins.pacman import Pacman from decman.plugins.systemd import Systemd @@ -44,6 +45,7 @@ execution_order: list[str] = [ pacman: None | Pacman = None aur: None | AUR = None systemd: None | Systemd = None +flatpak: None | Flatpak = None _pacman = plugins.get("pacman", None) if isinstance(_pacman, Pacman): @@ -57,6 +59,10 @@ _systemd = plugins.get("systemd", None) if isinstance(_systemd, Systemd): systemd = _systemd +_flatpak = plugins.get("flatpak", None) +if isinstance(_flatpak, Flatpak): + flatpak = _flatpak + def prg( cmd: list[str], diff --git a/src/decman/app.py b/src/decman/app.py index 166ddd2..ff9479a 100644 --- a/src/decman/app.py +++ b/src/decman/app.py @@ -46,7 +46,7 @@ def main(): help="don't run hook methods for modules", ) parser.add_argument( - "--params", nargs="*", type=str, help="additional parameters passed to pluging" + "--params", nargs="*", type=str, help="additional parameters passed to plugins" ) args = parser.parse_args() diff --git a/src/decman/core/command.py b/src/decman/core/command.py index 4f26307..fbdf012 100644 --- a/src/decman/core/command.py +++ b/src/decman/core/command.py @@ -79,7 +79,7 @@ def run( user: None | str = None, env_overrides: None | dict[str, str] = None, mimic_login: bool = False, - pass_environment: bool = False, + pass_environment: bool = True, ) -> tuple[int, str]: """ Runs a given command with the given arguments. The command can be ran as the given user and diff --git a/src/decman/plugins/aur/__init__.py b/src/decman/plugins/aur/__init__.py index b1064d0..d528011 100644 --- a/src/decman/plugins/aur/__init__.py +++ b/src/decman/plugins/aur/__init__.py @@ -22,12 +22,12 @@ __all__ = [ "AUR", "AurCommands", "CustomPackage", - "aur_packages", + "packages", "custom_packages", ] -def aur_packages(fn): +def packages(fn): """ Annotate that this function returns a set of AUR package names that should be installed. @@ -51,10 +51,10 @@ class AUR(plugins.Plugin): """ Plugin that manages additional pacman packages installed outside the pacman repos. - AUR packages are added directly to ``packages`` or declared by modules via ``@aur_packages``. + AUR packages are added directly to ``packages`` or declared by modules via ``@aur.packages``. Custom packages are added directly to ``custom_packages`` or declared by modules via - ``@custom_packages``. + ``@aur.custom_packages``. """ NAME = "aur" diff --git a/src/decman/plugins/flatpak.py b/src/decman/plugins/flatpak.py new file mode 100644 index 0000000..53964c8 --- /dev/null +++ b/src/decman/plugins/flatpak.py @@ -0,0 +1,258 @@ +import shutil + +import decman.core.command as command +import decman.core.error as errors +import decman.core.module as module +import decman.core.output as output +import decman.core.store as _store +import decman.plugins as plugins + + +def packages(fn): + """ + Annotate that this function returns a set of flatpak package names that should be installed. + + Return type of ``fn``: ``set[str]`` + """ + fn.__flatpak__packages__ = True + return fn + + +def user_packages(fn): + """ + Annotate that this function returns a dict of users and flatpak packages that should be + installed. + + Return type of ``fn``: ``dict[str, set[str]]`` + """ + fn.__flatpak__user__packages__ = True + return fn + + +class Flatpak(plugins.Plugin): + """ + Plugin that manages flatpak packages added directly to ``packages`` or declared by modules via + ``@flatpak.packages``. User packages are managed as well. + """ + + NAME = "flatpak" + + def __init__(self) -> None: + self.packages: set[str] = set() + self.user_packages: dict[str, set[str]] = {} + self.ignored_packages: set[str] = set() + self.commands = FlatpakCommands() + + def available(self) -> bool: + return shutil.which("flatpak") is not None + + def process_modules(self, store: _store.Store, modules: set[module.Module]): + # These store keys are used to track changes in modules. + # This way when these change, module can be marked as changed + store.ensure("flatpaks_for_module", {}) + store.ensure("user_flatpaks_for_module", {}) + + for mod in modules: + store["flatpaks_for_module"].setdefault(mod.name, set()) + store["user_flatpaks_for_module"].setdefault(mod.name, {}) + + packages = plugins.run_method_with_attribute(mod, "__flatpak__packages__") or set() + user_packages = ( + plugins.run_method_with_attribute(mod, "__flatpak__user__packages__") or {} + ) + + if store["flatpaks_for_module"][mod.name] != packages: + mod._changed = True + + if store["user_flatpaks_for_module"][mod.name] != user_packages: + mod._changed = True + + self.packages |= packages + for user, flatpaks in user_packages.items(): + self.user_packages.setdefault(user, set()).update(flatpaks) + + store["flatpaks_for_module"][mod.name] = packages + store["user_flatpaks_for_module"][mod.name] = user_packages + + def apply( + self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None + ) -> bool: + pm = FlatpakInterface(self.commands) + + try: + self.apply_packages(pm, None, self.packages, self.ignored_packages, dry_run) + + for user, packages in self.user_packages.items(): + self.apply_packages(pm, user, packages, self.ignored_packages, dry_run) + except errors.CommandFailedError as error: + output.print_error("Running a flatpak command failed.") + output.print_continuation(f"{error}") + output.print_traceback() + return False + return True + + def apply_packages( + self, + flatpak: "FlatpakInterface", + user: str | None, + packages: set[str], + ignored_packages: set[str], + dry_run: bool, + ): + currently_installed = flatpak.get_apps(user) + to_remove = currently_installed - packages - ignored_packages + to_install = packages - currently_installed - ignored_packages + + for_user_msg = f" for {user}" if user else "" + + if to_remove: + output.print_list(f"Removing flatpak packages{for_user_msg}:", sorted(to_remove)) + if not dry_run: + flatpak.remove(to_remove, user) + + output.print_summary(f"Upgrading packages{for_user_msg}.") + if not dry_run: + flatpak.upgrade(user) + + if to_install: + output.print_list(f"Installing flatpak packages{for_user_msg}:", sorted(to_install)) + if not dry_run: + flatpak.install(to_install, user) + + +class FlatpakCommands: + def list_apps(self, as_user: bool) -> list[str]: + """ + Running this command outputs a newline separated list of installed flatpak application IDs. + + If ``as_user`` is ``True``, run the command as the user whose packages should be listed. + + NOTE: The first line says 'Application ID' and should be ignored. + """ + return [ + "flatpak", + "list", + "--app", + "--user" if as_user else "--system", + "--columns", + "application", + ] + + def install(self, pkgs: set[str], as_user: bool) -> list[str]: + """ + Running this command installs all listed packages, and their dependencies/runtimes + automatically. + + If ``as_user`` is ``True``, run the command as the user for whom packages are installed. + """ + return [ + "flatpak", + "install", + "--user" if as_user else "--system", + ] + sorted(pkgs) + + def upgrade(self, as_user: bool) -> list[str]: + """ + Updates all installed flatpaks including runtimes and dependencies. + + If ``as_user`` is ``True``, run the command as the user whose flatpaks are updated. + """ + return [ + "flatpak", + "update", + "--user" if as_user else "--system", + ] + + def remove(self, pkgs: set[str], as_user: bool) -> list[str]: + """ + Running this command will remove the listed packages. + + If ``as_user`` is ``True``, run the command as the user for whom packages are removed. + """ + return [ + "flatpak", + "remove", + "--user" if as_user else "--system", + ] + sorted(pkgs) + + def remove_unused(self, as_user: bool) -> list[str]: + """ + This will remove all unused flatpak dependencies and runtimes. + + If ``as_user`` is ``True``, run the command as the user for whom packages are removed. + """ + return [ + "flatpak", + "remove", + "--unused", + "--user" if as_user else "--system", + ] + + +class FlatpakInterface: + """ + High level interface for running pacman commands. + + On failure methods raise a ``CommandFailedError``. + """ + + def __init__(self, commands: FlatpakCommands) -> None: + self._commands = commands + + def get_apps(self, user: str | None = None) -> set[str]: + """ + Returns a set of installed flatpak apps. + + If ``user`` is set, returns flatpak apps for that user. + """ + as_user = user is not None + + cmd = self._commands.list_apps(as_user=as_user) + _, packages_text = command.check_run_result( + cmd, command.run(cmd, user=user, mimic_login=as_user) + ) + packages = packages_text.strip().split("\n") + # Remove header "Application ID" + packages.pop(0) + + return set(packages) + + def install(self, packages: set[str], user: str | None = None): + """ + Installs the given packages. + + If ``user`` is set, installs packages for that user. + """ + if not packages: + return + + as_user = user is not None + + cmd = self._commands.install(packages, as_user) + command.check_run_result(cmd, command.pty_run(cmd, user=user, mimic_login=as_user)) + + def upgrade(self, user: str | None = None): + """ + Upgrades all packages. + + If ``user`` is set, upgrades packages for that user. + """ + as_user = user is not None + cmd = self._commands.upgrade(as_user) + command.check_run_result(cmd, command.pty_run(cmd, user=user, mimic_login=as_user)) + + def remove(self, packages: set[str], user: str | None = None): + """ + Removes the given packages as well as unused dependencies. + + If ``user`` is set, removes packages for that user. + """ + if not packages: + return + + as_user = user is not None + cmd = self._commands.remove(packages, as_user) + command.check_run_result(cmd, command.pty_run(cmd, user=user, mimic_login=as_user)) + + cmd = self._commands.remove_unused(as_user) + command.check_run_result(cmd, command.pty_run(cmd, user=user, mimic_login=as_user))