Added support for declaring installed and ignored flatpaks in the source file. For this the variables 'flatpak_packages' and 'ignored_flatpak_packages' are used. These work like the pacman equivalents. There were test cases appended for flatpak testing. I have installed the modified version on my system and tested it extensively. No errors were found.

2026-09-19 12:08:28 +00:00 · 2025-09-29 14:57:34 +02:00
parent f3910a6bc9
commit 401233a352
6 changed files with 235 additions and 59 deletions
@@ -412,3 +412,5 @@ enabled_systemd_user_units: dict[str, list[str]] = {}
 files: dict[str, File] = {}
 directories: dict[str, Directory] = {}
 modules: list[Module] = []
+flatpak_packages: list[str] = []
+ignored_flatpak_packages: list[str] = []
@@ -205,6 +205,7 @@ class Core:
        self.store = store
        self.source = _resolve_source()
        self.pacman = l.Pacman()
+        self.flatpak = l.Flatpak()
        self.systemctl = l.Systemd(store)
        self.fpkg_search = fpm.ExtendedPackageSearch(self.pacman)

@@ -257,36 +258,76 @@ class Core:
                self.systemctl.disable_user_units(units, user)

    def _remove_pkgs(self):
+        """
+        Remove pacman and flatpak packages
+        """
+        # pacman
        currently_installed = self.pacman.get_installed()
        to_remove = self.source.packages_to_remove(currently_installed)
-        l.print_list("Removing packages:", to_remove)
-        if not self.only_print:
-            self.pacman.remove(to_remove)
+
+        currently_installed_flatpak = self.flatpak.get_installed()
+        to_remove_flatpak = self.source.flatpak_packages_to_remove(
+            currently_installed_flatpak
+        )
+
+        l.print_list("Removing pacman packages:", to_remove)
+        l.print_list("Removing flatpak packages:", to_remove_flatpak)
+
+        if self.only_print:
+            return
+
+        self.pacman.remove(to_remove)
+
+        # flatpak
+        self.flatpak.remove(to_remove_flatpak)

    def _upgrade_pkgs(self):
+        """
+        Upgrade pacman, fpm and flatpak packages
+        """
+        # flatpak + fpm
        l.print_summary("Upgrading packages.")
-        if not self.only_print:
-            self.pacman.upgrade()
-            if conf.enable_fpm and self.update_foreign_packages:
-                self.fpm.upgrade(
-                    self.upgrade_devel, self.force_build, self.source.ignored_packages
-                )
+        if self.only_print:
+            return
+
+        self.pacman.upgrade()
+        if conf.enable_fpm and self.update_foreign_packages:
+            self.fpm.upgrade(
+                self.upgrade_devel, self.force_build, self.source.ignored_packages
+            )
+
+        # flatpak
+        self.flatpak.upgrade()

    def _install_pkgs(self):
+        """
+        Installs all pacman, fpm, and flatpak packages.
+        """
+
+        # pacman + fpm
        currently_installed = self.pacman.get_installed()
        to_install_pacman = self.source.pacman_packages_to_install(currently_installed)
        to_install_fpm = self.source.foreign_packages_to_install(currently_installed)

+        # flatpak
+        currently_installed_flatpak = self.flatpak.get_installed()
+        to_install_flatpak = self.source.flatpak_packages_to_install(
+            currently_installed_flatpak
+        )
+
        l.print_list("Installing pacman packages:", to_install_pacman)
+        l.print_list("Installing flatpak packages:", to_install_flatpak)

        # fpm prints a summary so no need to print it twice
        if self.only_print:
            l.print_list("Installing foreign packages:", to_install_fpm)
+            return

-        if not self.only_print:
-            self.pacman.install(to_install_pacman)
-            if conf.enable_fpm and self.update_foreign_packages:
-                self.fpm.install(to_install_fpm, force=self.force_build)
+        self.pacman.install(to_install_pacman)
+        if conf.enable_fpm and self.update_foreign_packages:
+            self.fpm.install(to_install_fpm, force=self.force_build)
+
+        self.flatpak.install(to_install_flatpak)

    def _create_and_remove_files(self):
        l.print_summary("Installing files.")
@@ -356,6 +397,8 @@ def _resolve_source() -> l.Source:
        files=decman.files,
        directories=decman.directories,
        modules=set(decman.modules),
+        flatpak_packages=set(decman.flatpak_packages),
+        ignored_flatpak_packages=set(decman.ignored_flatpak_packages),
    )


@@ -34,6 +34,13 @@ class Commands:
        """
        return ["pacman", "-Qeq", "--color=never"]

+    def list_flatpak_pkgs(self) -> list[str]:
+        """
+        Running this command outputs a newline separated list of installed flatpak application ids
+        The first line just says 'Application ID' so this one is ignored.
+        """
+        return ["flatpak", "list", "--app", "--columns", "application"]
+
    def list_foreign_pkgs_versioned(self) -> list[str]:
        """
        Running this command outputs a newline seperated list of installed packages and their
@@ -47,6 +54,12 @@ class Commands:
        """
        return ["pacman", "-S", "--color=always", "--needed"] + pkgs

+    def install_flatpak_pkgs(self, pkgs: list[str]) -> list[str]:
+        """
+        Running this command installs all listed packages, and their dependencies/runtimes automatically.
+        """
+        return ["flatpak", "install"] + pkgs
+
    def install_files(self, pkg_files: list[str]) -> list[str]:
        """
        Running this command installs the given packages files.
@@ -78,6 +91,12 @@ class Commands:
        """
        return ["pacman", "-Syu", "--color=always"]

+    def upgrade_flatpak(self) -> list[str]:
+        """
+        Updates all installed flatpak REFs including runtimes and dependencies.
+        """
+        return ["flatpak", "update"]
+
    def remove(self, pkgs: list[str]) -> list[str]:
        """
        Running this command removes the given packages and their dependencies
@@ -85,6 +104,18 @@ class Commands:
        """
        return ["pacman", "-Rs", "--color=always"] + pkgs

+    def remove_flatpak(self, pkgs: list[str]) -> list[str]:
+        """
+        Running this command will remove the listed REFs. Unused dependencies might be kept, but to remove them another command needs to be run.
+        """
+        return ["flatpak", "remove"] + pkgs
+
+    def remove_unused_flatpak(self) -> list[str]:
+        """
+        This will remove all unused flatpak dependencies and runtimes.
+        """
+        return ["flatpak", "remove", "--unused"]
+
    def enable_units(self, units: list[str]) -> list[str]:
        """
        Running this command enables the given systemd units.
@@ -419,6 +419,8 @@ class Source:
        files: dict[str, decman.File],
        directories: dict[str, decman.Directory],
        modules: set[decman.Module],
+        flatpak_packages: set[str],
+        ignored_flatpak_packages: set[str],
    ):
        self.pacman_packages = pacman_packages
        self.aur_packages = aur_packages
@@ -429,6 +431,8 @@ class Source:
        self.files = files
        self.directories = directories
        self.modules = modules
+        self.flatpak_packages = flatpak_packages
+        self.ignored_flatpak_packages = ignored_flatpak_packages

    def run_on_enable(self, store: Store):
        """
@@ -639,6 +643,36 @@ class Source:
                result.append(pkg)
        return result

+    def flatpak_packages_to_install(
+        self, currently_installed_packages: list[str]
+    ) -> list[str]:
+        """
+        Returns all flatpak packages, that are not installed or ignored
+        """
+
+        result: list[str] = []
+        for pkg in self.flatpak_packages:
+            if pkg in self.ignored_flatpak_packages:
+                continue
+            if pkg not in currently_installed_packages:
+                result.append(pkg)
+        return result
+
+    def flatpak_packages_to_remove(
+        self, currently_installed_packages: list[str]
+    ) -> list[str]:
+        """
+        This returns a list of flatpak app ids, that need to be removed since they are installed but not found in either the list of ignored packages or the list of flatpak packages that need to be installed.
+        """
+        result: list[str] = []
+        for package in currently_installed_packages:
+            if package in self.ignored_flatpak_packages:
+                continue
+            if package not in self.flatpak_packages:
+                result.append(package)
+
+        return result
+
    def all_enabled_modules(self) -> list[tuple[str, str]]:
        """
        Returns all enabled modules and their versions.
@@ -922,6 +956,87 @@ def echo_and_capture_command(program: list[str]) -> tuple[int, str]:
    return (returncode, output)


+class Flatpak:
+    def __init__(self) -> None:
+        pass
+
+    def get_installed(self) -> list[str]:
+        """
+        Return all of the installed applications. Dependencies and runtimes are exluded since they will not be explicitly installed and thus flatpak will manage them.
+        """
+        try:
+            packages = (
+                subprocess.run(
+                    conf.commands.list_flatpak_pkgs(),
+                    check=True,
+                    stdout=subprocess.PIPE,
+                )
+                .stdout.decode()
+                .strip()
+                .split("\n")
+            )
+
+            # The header might be included. It might also not. This will make sure that it is not present.
+            if "Application ID" in packages:
+                packages.remove("Application ID")
+
+            return packages
+        except subprocess.CalledProcessError as error:
+            raise err.UserFacingError(
+                user_facing_msg=f"Failed to get installed flatpak packages using '{error.cmd}'. Output: {error.stdout}."
+            ) from error
+
+    def install(self, packages: list[str]):
+        """
+        Install the listed flatpak packages.
+        """
+        if not packages:
+            return
+
+        returncode, _output = echo_and_capture_command(
+            conf.commands.install_flatpak_pkgs(packages)
+        )
+        if returncode != 0:
+            raise err.UserFacingError(
+                f"Failed to install flatpak packages. Process exited with code {returncode}."
+            )
+
+    def upgrade(self) -> None:
+        """
+        Upgrade all flatpak packages.
+        """
+        returncode, _output = echo_and_capture_command(conf.commands.upgrade_flatpak())
+        if not returncode == 0:
+            raise err.UserFacingError(
+                f"Failed to upgrade flatpak packages. Process exited with code {returncode}."
+            )
+
+    def remove(self, packages: list[str]):
+        """
+        Remove all the listed packages and their unused dependecies. This has to happen in two steps.
+        """
+        if not packages:
+            return
+
+        returncode, _output = echo_and_capture_command(
+            conf.commands.remove_flatpak(packages)
+        )
+
+        if not returncode == 0:
+            raise err.UserFacingError(
+                f"Failed to remove flatpak packages. Process exited with code {returncode}."
+            )
+
+        returncode, _output = echo_and_capture_command(
+            conf.commands.remove_unused_flatpak()
+        )
+
+        if not returncode == 0:
+            raise err.UserFacingError(
+                f"Failed to remove unused flatpak packages. Process exited with code {returncode}."
+            )
+
+
 class Systemd:
    """
    Interface for interacting with systemd.
@@ -6,7 +6,6 @@ from decman import UserPackage, Module


 class ExistingTestModule(Module):
-
    def __init__(self):
        self.on_enable_executed = False
        self.on_disable_executed = False
@@ -28,7 +27,6 @@ class ExistingTestModule(Module):


 class ExistingChangedVersionTestModule(Module):
-
    def __init__(self):
        self.on_enable_executed = False
        self.on_disable_executed = False
@@ -50,7 +48,6 @@ class ExistingChangedVersionTestModule(Module):


 class EnabledTestModule(Module):
-
    def __init__(self):
        self.on_enable_executed = False
        self.on_disable_executed = False
@@ -78,7 +75,6 @@ class EnabledTestModule(Module):


 class DisabledTestModule(Module):
-
    def __init__(self):
        self.on_enable_executed = False
        self.on_disable_executed = False
@@ -106,7 +102,6 @@ class DisabledTestModule(Module):


 class TestSource(unittest.TestCase):
-
    def setUp(self):
        self.disabled_module = DisabledTestModule()
        self.enabled_module = EnabledTestModule()
@@ -133,7 +128,7 @@ class TestSource(unittest.TestCase):
                    version="1",
                    dependencies=["d2"],
                    git_url="/am/url/yes",
-                )
+                ),
            },
            ignored_packages={"i1", "i2"},
            systemd_units={"1.service", "2.timer"},
@@ -141,11 +136,12 @@ class TestSource(unittest.TestCase):
            modules=modules,
            files={},
            directories={},
+            flatpak_packages={"f1", "f2", "f3"},
+            ignored_flatpak_packages={"i1", "i2"},
        )

        store = Store()
-        store.enabled_systemd_units.extend(
-            ["1.service", "3.service", "M_1.service"])
+        store.enabled_systemd_units.extend(["1.service", "3.service", "M_1.service"])
        store.add_enabled_user_systemd_unit("user", "u1.service")
        store.add_enabled_user_systemd_unit("user", "u3.service")
        store.enabled_modules = {
@@ -174,16 +170,19 @@ class TestSource(unittest.TestCase):
        self.currently_installed_packages = currently_installed_packages

    def test_all_enabled_modules(self):
-        enabled_modules = [("Enabled", "1"), ("Existing", "1"),
-                           ("ExistingChanged", "2")]
-        self.assertCountEqual(self.source.all_enabled_modules(),
-                              enabled_modules)
+        enabled_modules = [
+            ("Enabled", "1"),
+            ("Existing", "1"),
+            ("ExistingChanged", "2"),
+        ]
+        self.assertCountEqual(self.source.all_enabled_modules(), enabled_modules)

    def test_files_to_remove(self):
        created_files = ["/test/file1", "/test/file4"]
        self.assertCountEqual(
            self.source.files_to_remove(self.store, created_files),
-            ["/test/file2", "/test/file3"])
+            ["/test/file2", "/test/file3"],
+        )

    def test_after_update_executed(self):
        self.source.run_after_update()
@@ -197,8 +196,7 @@ class TestSource(unittest.TestCase):
        self.source.run_after_version_change(self.store)

        self.assertTrue(self.enabled_module.after_version_change_executed)
-        self.assertTrue(
-            self.existing_module_changed.after_version_change_executed)
+        self.assertTrue(self.existing_module_changed.after_version_change_executed)
        self.assertFalse(self.existing_module.after_version_change_executed)
        self.assertFalse(self.disabled_module.after_version_change_executed)

@@ -233,10 +231,7 @@ class TestSource(unittest.TestCase):
    def test_user_units_to_enable(self):
        self.assertDictEqual(
            self.source.user_units_to_enable(self.store),
-            {
-                "user": ["u2.timer"],
-                "muser": ["M_u1.service"]
-            },
+            {"user": ["u2.timer"], "muser": ["M_u1.service"]},
        )

    def test_user_units_to_disable(self):
@@ -247,15 +242,13 @@ class TestSource(unittest.TestCase):

    def test_pacman_packages_to_install(self):
        self.assertCountEqual(
-            self.source.pacman_packages_to_install(
-                self.currently_installed_packages),
+            self.source.pacman_packages_to_install(self.currently_installed_packages),
            ["p3", "M_p1", "M_p2"],
        )

    def test_foreign_packages_to_install(self):
        self.assertCountEqual(
-            self.source.foreign_packages_to_install(
-                self.currently_installed_packages),
+            self.source.foreign_packages_to_install(self.currently_installed_packages),
            ["A1", "U2"],
        )

@@ -265,46 +258,38 @@ class TestSource(unittest.TestCase):
            ["p4", "A4", "M_A1", "M_A2"],
        )

+
 class TestModuleUserServices(unittest.TestCase):
-
    class ModuleWithUserServiceOne(Module):
-
        def __init__(self):
            super().__init__("one", True, "0")

        def systemd_user_units(self) -> dict[str, list[str]]:
-            return {
-                    "user": ['foo.service']
-                    }
+            return {"user": ["foo.service"]}

    class ModuleWithUserServiceTwo(Module):
-
        def __init__(self):
            super().__init__("two", True, "0")

        def systemd_user_units(self) -> dict[str, list[str]]:
-            return {
-                    "user": ['bar.service']
-                    }
+            return {"user": ["bar.service"]}

    def setUp(self) -> None:
        self.source = Source(
-                pacman_packages=set(),
-                aur_packages=set(),
-                user_packages=set(),
-                ignored_packages=set(),
-                systemd_units=set(),
-                systemd_user_units={},
-                files={},
-                directories={},
-                modules={                
-                         self.ModuleWithUserServiceOne(),
-                         self.ModuleWithUserServiceTwo()
-                         },
-            )
+            pacman_packages=set(),
+            aur_packages=set(),
+            user_packages=set(),
+            ignored_packages=set(),
+            systemd_units=set(),
+            systemd_user_units={},
+            files={},
+            directories={},
+            modules={self.ModuleWithUserServiceOne(), self.ModuleWithUserServiceTwo()},
+            flatpak_packages=set(),
+            ignored_flatpak_packages=set(),
+        )
        self.store = Store()

-
    def test_user_units_to_enable(self):
        self.assertDictEqual(
            self.source.user_units_to_enable(self.store),