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

Switched the variable type of user defined flatpak packages to match the type of the user systemd services

Browse Source
This commit is contained in:
Dávid Groniewsky
2025-10-13 09:36:46 +02:00
parent 69bcaadf90
commit 6ae8c78959
7 changed files with 144 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dist/decman-0.3.4-py3-none-any.whl
Vendored
BIN
View File
Binary file not shown.
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[tuple[str,str]]:
return [("username", "io.github.kolunmi.Bazaar")]
def flatpak_user_packages(self) -> dict[str, list[str]]:
return {"username": ["io.github.kolunmi.Bazaar"]}
def systemd_units(self) -> list[str]:
return ["reflector.timer"]
example/source.py
+4 -3
View File
@@ -38,9 +38,10 @@ 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 += [
("username", "dev.zed.Zed")
]
# Ensure that previous user installed flatpak declarations aren't overwritten and they are initialized.
decman.flatpak_user_packages["kk"] = decman.flatpak_user_packages.get("kk", [])
# Now add the package.
decman.flatpak_user_packages["kk"].append("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
+3 -3
View File
@@ -387,12 +387,12 @@ class Module:
"""
return []
def flatpak_user_packages(self) -> list[tuple[str, str]]:
def flatpak_user_packages(self) -> dict[str, list[str]]:
"""
Override this method to return flatpak packages that should be installed to the user installation as a part of this
Module.
"""
return []
return {}
def systemd_units(self) -> list[str]:
"""
@@ -427,5 +427,5 @@ files: dict[str, File] = {}
directories: dict[str, Directory] = {}
modules: list[Module] = []
flatpak_packages: list[str] = []
flatpak_user_packages: list[tuple[str, str]] = []
flatpak_user_packages: dict[str, list[str]] = {}
ignored_flatpak_packages: list[str] = []
src/decman/app.py
+52 -14
View File
@@ -238,6 +238,7 @@ class Core:
"""
Run the main logic of decman.
"""
if self.update_units:
self._disable_units()
@@ -305,16 +306,33 @@ class Core:
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")
users = (
subprocess.run(["users"], check=True, stdout=subprocess.PIPE)
.stdout.decode()
.strip()
.split("\n")
)
print()
users.append("root")"""
users = pwd.getpwall()
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)
user = user.pw_name
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
if only_print:
continue
self.flatpak.remove(to_remove_flatpak, True, user)
@@ -373,16 +391,32 @@ class Core:
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")
"""# 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")"""
users = pwd.getpwall()
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)
user = user.pw_name
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
)
if only_print: continue
if only_print:
l.print_list(
f"Installing flatpak packages to user installation for user {user}",
to_install_flatpak,
)
continue
self.flatpak.install(to_install_flatpak, True, user)
@@ -444,6 +478,10 @@ def _resolve_source() -> l.Source:
for user, units in decman.enabled_systemd_user_units.items():
enabled_systemd_user_units[user] = set(units)
flatpak_user_packages = {}
for user, pkgs in decman.flatpak_user_packages.items():
flatpak_user_packages[user] = set(pkgs)
return l.Source(
pacman_packages=set(decman.packages),
aur_packages=set(decman.aur_packages),
@@ -455,7 +493,7 @@ def _resolve_source() -> l.Source:
directories=decman.directories,
modules=set(decman.modules),
flatpak_packages=set(decman.flatpak_packages),
flatpak_user_packages=set(decman.flatpak_user_packages),
flatpak_user_packages=flatpak_user_packages,
ignored_flatpak_packages=set(decman.ignored_flatpak_packages),
)
src/decman/config.py
+10 -4
View File
@@ -65,7 +65,7 @@ class Commands:
"""
Running this command installs all listed packages, and their dependencies/runtimes automatically.
"""
return ["flatpak", "install", "--user" if as_user else "--system"] + pkgs
return ["flatpak", "install", "-y", "--user" if as_user else "--system"] + pkgs
def install_files(self, pkg_files: list[str]) -> list[str]:
"""
@@ -102,7 +102,7 @@ class Commands:
"""
Updates all installed flatpak REFs including runtimes and dependencies.
"""
return ["flatpak", "update", "--user" if as_user else "--system"]
return ["flatpak", "update", "-y", "--user" if as_user else "--system"]
def remove(self, pkgs: list[str]) -> list[str]:
"""
@@ -115,13 +115,19 @@ class Commands:
"""
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", "--user" if as_user else "--system"] + pkgs
return ["flatpak", "remove", "-y", "--user" if as_user else "--system"] + pkgs
def remove_unused_flatpak(self, as_user: bool = False) -> list[str]:
"""
This will remove all unused flatpak dependencies and runtimes.
"""
return ["flatpak", "remove", "--unused", "--user" if as_user else "--system"]
return [
"flatpak",
"remove",
"-y",
"--unused",
"--user" if as_user else "--system",
]
def enable_units(self, units: list[str]) -> list[str]:
"""
src/decman/lib/__init__.py
+73 -44
View File
@@ -36,6 +36,7 @@ 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.
@@ -43,6 +44,7 @@ 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.
@@ -50,6 +52,7 @@ 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.
@@ -57,6 +60,7 @@ def print_summary(msg: str):
print(f"{_DECMAN_MSG_TAG} {_CYAN_PREFIX}SUMMARY{_RESET_SUFFIX}: {msg}")
def print_list(
msg: str,
l: list[str],
@@ -116,6 +120,7 @@ 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.
@@ -123,6 +128,7 @@ 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.
@@ -130,6 +136,7 @@ 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:
@@ -412,7 +419,7 @@ class Source:
directories: dict[str, decman.Directory],
modules: set[decman.Module],
flatpak_packages: set[str],
flatpak_user_packages: set[tuple[str,str]],
flatpak_user_packages: dict[str, set[str]],
ignored_flatpak_packages: set[str],
):
self.pacman_packages = pacman_packages
@@ -638,7 +645,10 @@ class Source:
return result
def flatpak_packages_to_install(
self, currently_installed_packages: list[str], as_user: bool = False, which_user: str = ""
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
@@ -653,7 +663,10 @@ class Source:
return result
def flatpak_packages_to_remove(
self, currently_installed_packages: list[str], as_user: bool = False, which_user: str = ""
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,
@@ -697,29 +710,24 @@ class Source:
result.update(module.pacman_packages())
return result
def _all_flatpak_packages(self, as_user: bool = False, which_user: str = "") -> 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])
current_user_flatpak_packages = self.flatpak_user_packages.get(which_user, [])
result = set()
result.update(
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_current_user_flatpak_packages
)
if not module.enabled:
continue
result.update(
module.flatpak_packages()
if not as_user
else module.flatpak_user_packages().get(which_user, [])
)
return result
@@ -956,6 +964,7 @@ 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.
@@ -975,10 +984,12 @@ 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
@@ -988,19 +999,28 @@ class Flatpak:
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()
.split("\n")
uinfo: tuple[int, int] = (0, 0)
env = os.environ.copy()
user_env = env.copy()
user_env["HOME"] = os.path.expanduser(f"~{which_user}")
if as_user:
uinfo = get_user_info(which_user)
proc = subprocess.run(
conf.commands.list_flatpak_pkgs(as_user),
check=True,
stdout=subprocess.PIPE,
user=uinfo[0],
group=uinfo[1],
env=user_env if as_user else env,
)
packages = proc.stdout.decode().strip().split("\n")
# print(
# f"as_user: {as_user}, which_user: {which_user}, uinfo: {uinfo}, stdout: {proc.stdout.decode()}, packages: {packages}"
# )
# The header might be included. It might also not. This will make sure that it is not present.
if "Application ID" in packages:
@@ -1015,14 +1035,17 @@ 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, which_user: str = ""):
def install(self, packages: list[str], as_user: bool = False, which_user: str = ""):
"""
Install the listed flatpak packages.
"""
if not packages:
return
uinfo = get_user_info(which_user)
uinfo: tuple[int, int] = (0, 0)
if as_user:
uinfo = get_user_info(which_user)
env = os.environ.copy()
user_env = env.copy()
user_env["HOME"] = os.path.expanduser(f"~{which_user}")
@@ -1031,9 +1054,9 @@ class Flatpak:
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,
env=user_env if as_user else env
user=uinfo[0],
group=uinfo[1],
env=user_env if as_user else env,
)
if proc.returncode != 0:
@@ -1045,10 +1068,12 @@ class Flatpak:
"""
Upgrade all flatpak packages.
"""
returncode, _output = echo_and_capture_command(conf.commands.upgrade_flatpak())
if not returncode == 0:
proc = subprocess.run(
conf.commands.upgrade_flatpak(), check=True, stdout=subprocess.PIPE
)
if not proc.returncode == 0:
raise err.UserFacingError(
f"Failed to upgrade flatpak packages. Process exited with code {returncode}."
f"Failed to upgrade flatpak packages. Process exited with code {proc.returncode}."
)
def remove(self, packages: list[str], as_user: bool = False, which_user: str = ""):
@@ -1058,7 +1083,10 @@ class Flatpak:
if not packages:
return
uinfo = get_user_info(which_user)
uinfo: tuple[int, int] = (0, 0)
if as_user:
uinfo = get_user_info(which_user)
env = os.environ.copy()
user_env = env.copy()
user_env["HOME"] = os.path.expanduser(f"~{which_user}")
@@ -1067,9 +1095,9 @@ class Flatpak:
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,
env=user_env if as_user else env
user=uinfo[0],
group=uinfo[1],
env=user_env if as_user else env,
)
if not proc.returncode == 0:
@@ -1083,7 +1111,7 @@ class Flatpak:
stdout=subprocess.PIPE,
user=uinfo[0] if as_user else 0,
group=uinfo[1] if as_user else 0,
env=user_env if as_user else env
env=user_env if as_user else env,
)
if not proc.returncode == 0:
@@ -1091,6 +1119,7 @@ class Flatpak:
f"Failed to remove unused flatpak packages. Process exited with code {proc.returncode}."
)
class Systemd:
"""
Interface for interacting with systemd.