JustAnyone/decman
1
0
Fork 0
mirror of https://github.com/kiviktnm/decman.git synced 2026-09-19 12:08:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fixed the management of flatpak packages installed to the user installation as requested in the PR

Browse Source
This commit is contained in:
David Groniewsky
2025-10-09 22:42:19 +02:00
parent d765eea3f4
commit 7ea1ef2d81
5 changed files with 99 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
example/my_module.py
+2 -2
View File
@@ -116,8 +116,8 @@ class MyModule(Module):
def flatpak_packages(self) -> list[str]:
return ["org.mozilla.firefox"]
def flatpak_user_packages(self) -> list[str]:
return ["io.github.kolunmi.Bazaar"]
def flatpak_user_packages(self) -> list[tuple[str,str]]:
return [("username", "io.github.kolunmi.Bazaar")]
def systemd_units(self) -> list[str]:
return ["reflector.timer"]
example/source.py
+3 -1
View File
@@ -38,7 +38,9 @@ decman.flatpak_packages += ["dev.qwery.AddWater"]
decman.ignored_flatpak_packages += ["org.signal.Signal"]
# You can also install them to your user installation instead of the system installation
decman.flatpak_user_packages += ["dev.zed.Zed"]
decman.flatpak_user_packages += [
("username", "dev.zed.Zed")
]
# To import GPG keys, set the GNUPGHOME environment variable.
# It can easily be done with python as well.
src/decman/__init__.py
+2 -2
View File
@@ -387,7 +387,7 @@ class Module:
"""
return []
def flatpak_user_packages(self) -> list[str]:
def flatpak_user_packages(self) -> list[tuple[str, str]]:
"""
Override this method to return flatpak packages that should be installed to the user installation as a part of this
Module.
@@ -427,5 +427,5 @@ files: dict[str, File] = {}
directories: dict[str, Directory] = {}
modules: list[Module] = []
flatpak_packages: list[str] = []
flatpak_user_packages: list[str] = []
flatpak_user_packages: list[tuple[str, str]] = []
ignored_flatpak_packages: list[str] = []
src/decman/app.py
+34 -12
View File
@@ -6,8 +6,10 @@ Module containing the CLI Application.
import argparse
import os
import shutil
import subprocess
import sys
import traceback
import pwd
import decman
import decman.config as conf
@@ -285,16 +287,12 @@ class Core:
to_remove_flatpak = self.source.flatpak_packages_to_remove(
currently_installed_flatpak
)
currently_installed_user_flatpak = self.flatpak.get_installed(True)
to_remove_user_flatpak = self.source.flatpak_packages_to_remove(
currently_installed_user_flatpak, as_user=True
)
l.print_list("Removing pacman packages:", to_remove)
if conf.enable_flatpak:
l.print_list("Removing flatpak packages:", to_remove_flatpak)
l.print_list("Removing user flatpak packages:", to_remove_user_flatpak)
self._remove_user_flatpaks(only_print=True)
if self.only_print:
return
@@ -304,7 +302,21 @@ class Core:
# flatpak
if conf.enable_flatpak and self.update_flatpaks:
self.flatpak.remove(to_remove_flatpak)
self.flatpak.remove(to_remove_user_flatpak, True)
self._remove_user_flatpaks()
def _remove_user_flatpaks(self, only_print: bool = False):
# get all users through a command instead of pwd because pwd also lists all 'virtual' users. add root since they can also have user installed packages
users = subprocess.run(["users"], check=True, stdout=subprocess.PIPE).stdout.decode().strip().split('\n')
users.append("root")
for user in users:
currently_installed_flatpak = self.flatpak.get_installed(as_user=True, which_user=user)
to_remove_flatpak = self.source.flatpak_packages_to_remove(currently_installed_flatpak, as_user=True, which_user=user)
l.print_list(f"Removing flatpak packages from user installation for user {user}", to_remove_flatpak)
if only_print: continue
self.flatpak.remove(to_remove_flatpak, True, user)
def _upgrade_pkgs(self):
"""
@@ -340,16 +352,12 @@ class Core:
to_install_flatpak = self.source.flatpak_packages_to_install(
currently_installed_flatpak
)
currently_installed_user_flatpak = self.flatpak.get_installed(True)
to_install_user_flatpak = self.source.flatpak_packages_to_install(
currently_installed_user_flatpak, True
)
l.print_list("Installing pacman packages:", to_install_pacman)
if conf.enable_flatpak:
l.print_list("Installing flatpak packages:", to_install_flatpak)
l.print_list("Installing user flatpak packages:", to_install_user_flatpak)
self._install_user_flatpaks(only_print=True)
# fpm prints a summary so no need to print it twice
if self.only_print:
@@ -362,7 +370,21 @@ class Core:
if conf.enable_flatpak and self.update_flatpaks:
self.flatpak.install(to_install_flatpak)
self.flatpak.install(to_install_user_flatpak, True)
self._install_user_flatpaks()
def _install_user_flatpaks(self, only_print: bool = False):
# get all users through a command instead of pwd because pwd also lists all 'virtual' users. add root since they can also have user installed packages
users = subprocess.run(["users"], check=True, stdout=subprocess.PIPE).stdout.decode().strip().split('\n')
users.append("root")
for user in users:
currently_installed_flatpak = self.flatpak.get_installed(as_user=True, which_user=user)
to_install_flatpak = self.source.flatpak_packages_to_install(currently_installed_flatpak, as_user=True, which_user=user)
l.print_list(f"Installing flatpak packages to user installation for user {user}", to_install_flatpak)
if only_print: continue
self.flatpak.install(to_install_flatpak, True, user)
def _create_and_remove_files(self):
l.print_summary("Installing files.")
src/decman/lib/__init__.py
+58 -34
View File
@@ -7,10 +7,9 @@ import os
import pty
import shutil
import subprocess
import sys
import threading
import time
import typing
import pwd
import decman
import decman.config as conf
@@ -37,7 +36,6 @@ def print_continuation(msg: str, level: int = SUMMARY):
if level == SUMMARY or conf.debug_output or not conf.quiet_output:
print(f"{_CONTINUATION_PREFIX}{msg}")
def print_error(error_msg: str):
"""
Prints an error message to the user.
@@ -45,7 +43,6 @@ def print_error(error_msg: str):
print(f"{_DECMAN_MSG_TAG} {_RED_PREFIX}ERROR{_RESET_SUFFIX}: {error_msg}")
def print_warning(msg: str):
"""
Prints a warning to the user.
@@ -53,7 +50,6 @@ def print_warning(msg: str):
print(f"{_DECMAN_MSG_TAG} {_YELLOW_PREFIX}WARNING{_RESET_SUFFIX}: {msg}")
def print_summary(msg: str):
"""
Prints a summary message to the user.
@@ -61,7 +57,6 @@ def print_summary(msg: str):
print(f"{_DECMAN_MSG_TAG} {_CYAN_PREFIX}SUMMARY{_RESET_SUFFIX}: {msg}")
def print_list(
msg: str,
l: list[str],
@@ -121,7 +116,6 @@ def print_list(
print_continuation("", level=level)
def print_info(msg: str):
"""
Prints a detailed message to the user if verbose output is not disabled.
@@ -129,7 +123,6 @@ def print_info(msg: str):
if conf.debug_output or not conf.quiet_output:
print(f"{_DECMAN_MSG_TAG} INFO: {msg}")
def print_debug(msg: str):
"""
Prints a detailed message to the user if debug messages are enabled.
@@ -137,7 +130,6 @@ def print_debug(msg: str):
if conf.debug_output:
print(f"{_DECMAN_MSG_TAG} {_GRAY_PREFIX}DEBUG{_RESET_SUFFIX}: {msg}")
def prompt_number(
msg: str, min_num: int, max_num: int, default: typing.Optional[int] = None
) -> int:
@@ -420,7 +412,7 @@ class Source:
directories: dict[str, decman.Directory],
modules: set[decman.Module],
flatpak_packages: set[str],
flatpak_user_packages: set[str],
flatpak_user_packages: set[tuple[str,str]],
ignored_flatpak_packages: set[str],
):
self.pacman_packages = pacman_packages
@@ -646,14 +638,14 @@ class Source:
return result
def flatpak_packages_to_install(
self, currently_installed_packages: list[str], as_user: bool = False
self, currently_installed_packages: list[str], as_user: bool = False, which_user: str = ""
) -> list[str]:
"""
Returns all flatpak packages, that are not installed or ignored
"""
result: list[str] = []
for pkg in self._all_flatpak_packages(as_user):
for pkg in self._all_flatpak_packages(as_user, which_user):
if pkg in self.ignored_flatpak_packages:
continue
if pkg not in currently_installed_packages:
@@ -661,7 +653,7 @@ class Source:
return result
def flatpak_packages_to_remove(
self, currently_installed_packages: list[str], as_user: bool = False
self, currently_installed_packages: list[str], as_user: bool = False, which_user: 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,
@@ -671,7 +663,7 @@ class Source:
for package in currently_installed_packages:
if package in self.ignored_flatpak_packages:
continue
if package not in self._all_flatpak_packages(as_user):
if package not in self._all_flatpak_packages(as_user, which_user):
result.append(package)
return result
@@ -705,17 +697,28 @@ class Source:
result.update(module.pacman_packages())
return result
def _all_flatpak_packages(self, as_user: bool = False) -> set[str]:
def _all_flatpak_packages(self, as_user: bool = False, which_user: str = "") -> set[str]:
# loop through all the user packages and save which ones are owned by the currently selected user
current_user_flatpak_packages = []
for pkg in self.flatpak_user_packages:
if not pkg[0] == which_user: continue
current_user_flatpak_packages.append(pkg[1])
result = set()
result.update(
self.flatpak_packages if not as_user else self.flatpak_user_packages
self.flatpak_packages if not as_user else current_user_flatpak_packages
)
for module in self.modules:
if module.enabled:
module_current_user_flatpak_packages = []
for pkg in module.flatpak_user_packages():
if not pkg[0] == which_user: continue
module_current_user_flatpak_packages.append(pkg[1])
result.update(
module.flatpak_packages()
if not as_user
else module.flatpak_user_packages()
else module_current_user_flatpak_packages
)
return result
@@ -953,7 +956,6 @@ def print_highlighted_pacman_messages(output: str):
# Break, as to not print the same line again if it contains multiple keywords
break
def echo_and_capture_command(program: list[str]) -> tuple[int, str]:
"""
Runs the given CLI program and arguments.
@@ -973,21 +975,27 @@ def echo_and_capture_command(program: list[str]) -> tuple[int, str]:
return (returncode, output)
def get_user_info(username: str) -> tuple[int, int]:
info = pwd.getpwnam(username)
return (info.pw_uid, info.pw_gid)
class Flatpak:
def __init__(self) -> None:
pass
def get_installed(self, as_user: bool = False) -> list[str]:
def get_installed(self, as_user: bool = False, which_user: str = "") -> 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:
uinfo = get_user_info(which_user)
packages = (
subprocess.run(
conf.commands.list_flatpak_pkgs(as_user),
check=True,
stdout=subprocess.PIPE,
user=uinfo[0] if as_user else 0,
group=uinfo[1] if as_user else 0,
)
.stdout.decode()
.strip()
@@ -1007,19 +1015,26 @@ class Flatpak:
user_facing_msg=f"Failed to get installed flatpak packages using '{error.cmd}'. Output: {error.stdout}."
) from error
def install(self, packages: list[str], as_user: bool = False):
def install(self, packages: list[str], as_user: bool = False, which_user: str = ""):
"""
Install the listed flatpak packages.
"""
if not packages:
return
returncode, _output = echo_and_capture_command(
conf.commands.install_flatpak_pkgs(packages, as_user)
uinfo = get_user_info(which_user)
proc = subprocess.run(
conf.commands.install_flatpak_pkgs(packages, as_user),
check=True,
stdout=subprocess.PIPE,
user=uinfo[0] if as_user else 0,
group=uinfo[1] if as_user else 0,
)
if returncode != 0:
if proc.returncode != 0:
raise err.UserFacingError(
f"Failed to install flatpak packages. Process exited with code {returncode}."
f"Failed to install flatpak packages. Process exited with code {proc.returncode}."
)
def upgrade(self) -> None:
@@ -1032,32 +1047,41 @@ class Flatpak:
f"Failed to upgrade flatpak packages. Process exited with code {returncode}."
)
def remove(self, packages: list[str], as_user: bool = False):
def remove(self, packages: list[str], as_user: bool = False, which_user: 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, as_user)
uinfo = get_user_info(which_user)
proc = subprocess.run(
conf.commands.remove_flatpak(packages, as_user),
check=True,
stdout=subprocess.PIPE,
user=uinfo[0] if as_user else 0,
group=uinfo[1] if as_user else 0,
)
if not returncode == 0:
if not proc.returncode == 0:
raise err.UserFacingError(
f"Failed to remove flatpak packages. Process exited with code {returncode}."
f"Failed to remove flatpak packages. Process exited with code {proc.returncode}."
)
returncode, _output = echo_and_capture_command(
conf.commands.remove_unused_flatpak()
proc = subprocess.run(
conf.commands.remove_unused_flatpak(as_user),
check=True,
stdout=subprocess.PIPE,
user=uinfo[0] if as_user else 0,
group=uinfo[1] if as_user else 0,
)
if not returncode == 0:
if not proc.returncode == 0:
raise err.UserFacingError(
f"Failed to remove unused flatpak packages. Process exited with code {returncode}."
f"Failed to remove unused flatpak packages. Process exited with code {proc.returncode}."
)
class Systemd:
"""
Interface for interacting with systemd.