diff --git a/example/my_module.py b/example/my_module.py new file mode 100644 index 0000000..155eaf1 --- /dev/null +++ b/example/my_module.py @@ -0,0 +1,19 @@ +# from import is ok for importing classes and functions +# just remember to not import variables this way +from decman import Module, sh + +import decman + + +class MyModule(Module): + + def __init__(self): + self.pkgs = ["rust"] + super().__init__("Example module", True, "1") + + def enable_my_custom_feature(self, b: bool): + if b: + self.pkgs = ["rustup"] + + def pacman_packages(self) -> list[str]: + return self.pkgs diff --git a/example/source.py b/example/source.py new file mode 100644 index 0000000..5d6e2b4 --- /dev/null +++ b/example/source.py @@ -0,0 +1,9 @@ +import decman +from my_module import MyModule + +my_own_mod = MyModule() + +my_own_mod.enable_my_custom_feature(True) + +decman.packages += ["python", "python"] +decman.modules += [my_own_mod] diff --git a/pyproject.toml b/pyproject.toml index d358ea5..d9f8b5b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "decman" version = "0.0.1" -description = "Declarative package/configuration manager for Arch Linux" +description = "Declarative package & configuration manager for Arch Linux" authors = [ {name = "Kivi Kaitaniemi"} ] diff --git a/src/decman/__init__.py b/src/decman/__init__.py index d6845e6..ca79bac 100644 --- a/src/decman/__init__.py +++ b/src/decman/__init__.py @@ -208,8 +208,8 @@ class Directory: original_wd = os.getcwd() try: os.chdir(self.source_directory) - for src_dir, _, files in os.walk("."): - for src_file in files: + for src_dir, _, src_files in os.walk("."): + for src_file in src_files: src_path = os.path.join(src_dir, src_file) file = File(source_file=src_path, bin_file=self.bin_files, @@ -259,6 +259,14 @@ class UserPackage: self.check_dependencies = check_dependencies self.git_url = git_url + def __hash__(self) -> int: + return self.pkgname.__hash__() + + def __eq__(self, value: object, /) -> bool: + if isinstance(value, self.__class__): + return value.pkgname == self.pkgname + return False + class Module: """ @@ -349,3 +357,22 @@ class Module: Module. """ return {} + + def __hash__(self) -> int: + return self.name.__hash__() + + def __eq__(self, value: object, /) -> bool: + if isinstance(value, self.__class__): + return value.name == self.name + return False + + +packages: list[str] = [] +aur_packages: list[str] = [] +user_packages: list[UserPackage] = [] +ignored_packages: list[str] = [] +enabled_systemd_units: list[str] = [] +enabled_systemd_user_units: dict[str, list[str]] = {} +files: dict[str, File] = {} +directories: dict[str, Directory] = {} +modules: list[Module] = [] diff --git a/src/decman/__main__.py b/src/decman/__main__.py new file mode 100644 index 0000000..8175ca8 --- /dev/null +++ b/src/decman/__main__.py @@ -0,0 +1,3 @@ +import decman.app + +decman.app.main() diff --git a/src/decman/app.py b/src/decman/app.py index 891c53c..1de33c4 100644 --- a/src/decman/app.py +++ b/src/decman/app.py @@ -2,10 +2,15 @@ Module containing the CLI Application. """ +import argparse import os import sys +import traceback -from decman.lib import AUR, Pacman, Systemd, Store, print_error +import decman +import decman.error as err +import decman.lib as l +import decman.lib.aur as aur def main(): @@ -13,26 +18,262 @@ def main(): Main entry for the CLI app """ - aur = AUR() - - print(aur.get_package_info("zapzap")) - - if not is_root(): - print_error("Not running as root. Please run decman as root.") + if not _is_root(): + l.print_error("Not running as root. Please run decman as root.") sys.exit(1) - p = Pacman() + original_wd = os.getcwd() - print(p.get_versioned_foreign_packages()) + try: + store = l.Store.restore() + opts = _set_up(store) + Core(store, opts).run() + except err.UserFacingError as error: + l.print_error(error.user_facing_msg) + for line in traceback.format_exc().splitlines(): + l.print_debug(line) + sys.exit(1) + + # Save even when an error has occurred, since this avoids repeating steps like building pkgs. + try: + store.save() + except err.UserFacingError as error: + l.print_error(error.user_facing_msg) + for line in traceback.format_exc().splitlines(): + l.print_debug(line) + sys.exit(1) + + os.chdir(original_wd) -def is_root() -> bool: - """ - Returns True if the process is running as root. - """ +def _set_up(store: l.Store): + parser = argparse.ArgumentParser( + prog="decman", + description= + "Declarative package & configuration manager for Arch Linux", + epilog="See more help at: https://github.com/kiviktnm/decman") + parser.add_argument("--source", + action="store", + help="python file containing configuration") + parser.add_argument( + "--print", + action="store_true", + default=False, + help= + "print what would happen as a result of running decman (doesn't print removed files)" + ) + parser.add_argument("--no-packages", + action="store_true", + default=False, + help="don't upgrade any packages") + parser.add_argument("--no-files", + action="store_true", + default=False, + help="don't install any files") + parser.add_argument("--no-systemd-units", + action="store_true", + default=False, + help="don't enable/disable systemd units") + parser.add_argument("--upgrade-devel", + action="store_true", + default=False, + help="upgrade devel packages") + parser.add_argument( + "--force-build", + action="store_true", + default=False, + help="force building of packages that are already cached") + + args = parser.parse_args() + source = store.source_file + source_changed = False + if args.source is not None: + source = args.source + source_changed = True + + if source is None: + l.print_error( + "Source was not specified. Please specify a source at least once with the --source argument." + ) + l.print_info("Decman will remember the previous source.") + sys.exit(1) + + if source_changed or not store.allow_running_source_without_prompt: + l.print_warning(f"Decman will run the file '{source}' as root!") + l.print_warning( + "Only proceed if you trust the file completely. The file can also import other files." + ) + + if not l.prompt_confirm("Proceed?", default=False): + sys.exit(1) + + if l.prompt_confirm("Remember this choice?", default=False): + store.allow_running_source_without_prompt = True + + source_path = os.path.abspath(source) + source_dir = os.path.dirname(source_path) + store.source_file = source_path + + try: + with open(source_path, "rt", encoding="utf-8") as file: + content = file.read() + except OSError as e: + raise err.UserFacingError( + f"Failed to read source file '{store.source_file}'.") from e + + os.chdir(source_dir) + sys.path.append(".") + exec(content) + + return args.print, not args.no_packages, not args.no_files, not args.no_systemd_units, args.upgrade_devel, args.force_build + + +class Core: + + def __init__(self, store: l.Store, opts): + self.only_print, self.update_packages, self.update_files, self.update_units, self.upgrade_devel, self.force_build = opts + + self.store = store + self.source = _resolve_source() + self.pacman = l.Pacman() + self.systemctl = l.Systemd(store) + self.fpkg_search = aur.ExtendedPackageSearch(self.pacman) + + for upkg in self.source.user_packages: + self.fpkg_search.add_user_pkg( + aur.PackageInfo.from_user_package(upkg, self.pacman)) + + self.fpm = aur.ForeignPackageManager(store, self.pacman, + self.fpkg_search) + + def run(self): + if self.update_units: + self._disable_units() + + if self.update_packages: + self._remove_pkgs() + self._upgrade_pkgs() + self._install_pkgs() + + if self.update_files: + self._create_and_remove_files() + + if self.update_units: + self._enable_units() + + self._run_modules() + + def _disable_units(self): + to_disable = self.source.units_to_disable(self.store) + l.print_list_summary("Disabling systemd units:", to_disable) + if not self.only_print: + self.systemctl.disable_units(to_disable) + + user_units_to_disable = self.source.user_units_to_disable(self.store) + for user, units in user_units_to_disable.items(): + l.print_list_summary(f"Disabling systemd units for {user}:", units) + if not self.only_print: + self.systemctl.disable_user_units(units, user) + + def _remove_pkgs(self): + currently_installed = self.pacman.get_installed() + to_remove = self.source.packages_to_remove(currently_installed) + l.print_list_summary("Removing packages:", to_remove) + if not self.only_print: + self.pacman.remove(to_remove) + + def _upgrade_pkgs(self): + l.print_summary("Upgrading packages.") + if not self.only_print: + self.pacman.upgrade() + self.fpm.upgrade(self.upgrade_devel, self.force_build, + self.source.ignored_packages) + + def _install_pkgs(self): + 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) + + l.print_list_summary("Installing pacman packages:", to_install_pacman) + l.print_list_summary("Installing foreign packages:", to_install_fpm) + + if not self.only_print: + self.pacman.install(to_install_pacman) + self.fpm.install(to_install_fpm, force=self.force_build) + + def _create_and_remove_files(self): + l.print_list_summary("Copying files:", + self.source.all_file_targets(), + elements_per_line=1) + l.print_list_summary("Copying directories:", + self.source.all_directory_targets(), + elements_per_line=1) + + if self.only_print: + return + + all_created = self.source.create_all_files() + to_remove = self.source.files_to_remove(self.store, all_created) + + l.print_list_summary("Removing files:", to_remove, elements_per_line=1) + + for file in to_remove: + try: + os.remove(file) + except OSError as e: + l.print_error(f"{e}") + l.print_warning(f"Failed to remove file: {file}") + + def _enable_units(self): + to_enable = self.source.units_to_enable(self.store) + l.print_list_summary("Enabling systemd units:", to_enable) + if not self.only_print: + self.systemctl.enable_units(to_enable) + + user_units_to_enable = self.source.user_units_to_enable(self.store) + for user, units in user_units_to_enable.items(): + l.print_list_summary(f"Enabling systemd units for {user}:", units) + if not self.only_print: + self.systemctl.enable_user_units(units, user) + + def _run_modules(self): + l.print_summary("Running on enable hooks.") + if not self.only_print: + self.source.run_on_enable(self.store) + + l.print_summary("Running after version change hooks.") + if not self.only_print: + self.source.run_after_version_change(self.store) + + l.print_summary("Running on disable hooks.") + if not self.only_print: + self.source.run_on_disable(self.store) + + l.print_summary("Running after update hooks.") + if not self.only_print: + self.source.run_after_update() + + +def _resolve_source() -> l.Source: + enabled_systemd_user_units = {} + for user, units in decman.enabled_systemd_user_units: + enabled_systemd_user_units[user] = set(units) + + return l.Source( + pacman_packages=set(decman.packages), + aur_packages=set(decman.aur_packages), + user_packages=set(decman.user_packages), + ignored_packages=set(decman.ignored_packages), + systemd_units=set(decman.enabled_systemd_units), + systemd_user_units=enabled_systemd_user_units, + files=decman.files, + directories=decman.directories, + modules=set(decman.modules), + ) + + +def _is_root() -> bool: return os.geteuid() == 0 - - -if __name__ == "__main__": - main() diff --git a/src/decman/lib/__init__.py b/src/decman/lib/__init__.py index 8920ec0..be5d480 100644 --- a/src/decman/lib/__init__.py +++ b/src/decman/lib/__init__.py @@ -51,6 +51,31 @@ def print_summary(msg: str): print(f"{_DECMAN_MSG_TAG} {_CYAN_PREFIX}SUMMARY{_RESET_SUFFIX}: {msg}") +def print_list_summary(msg: str, + l: list[str], + elements_per_line: typing.Optional[int] = None): + """ + Prints a summary message to the user along with a list of elements. + + If the list is empty, prints nothing. + """ + if len(l) == 0: + return + + l = l.copy() + print_summary(msg) + print_continuation("") + if elements_per_line is None: + print_continuation(" ".join(l)) + else: + while l: + to_print = [] + for _ in range(elements_per_line): + to_print.append(l.pop()) + print_continuation(" ".join(to_print)) + print_continuation("") + + def print_info(msg: str): """ Prints a detailed message to the user if verbose output is not disabled. @@ -132,6 +157,8 @@ class Store: """ def __init__(self): + self.source_file: typing.Optional[str] = None + self.allow_running_source_without_prompt: bool = False self.enabled_systemd_units: list[str] = [] self.enabled_user_systemd_units: list[tuple[str, str]] = [] self.enabled_modules: dict[str, str] = {} @@ -169,6 +196,9 @@ class Store: print_debug(f"Writing Store to '{path}'.") d = { + "source_file": self.source_file, + "allow_running_source_without_prompt": + self.allow_running_source_without_prompt, "enabled_systemd_units": self.enabled_systemd_units, "enabled_user_systemd_units": self.enabled_user_systemd_units, "enabled_modules": self.enabled_modules, @@ -203,6 +233,9 @@ class Store: with open(path, "rt", encoding="utf-8") as file: d = json.load(file) + store.source_file = d.get("source_file", None) + store.allow_running_source_without_prompt = d.get( + "allow_running_source_without_prompt", False) store.enabled_systemd_units = d.get( "enabled_systemd_units", [], @@ -237,15 +270,15 @@ class Source: def __init__( self, - pacman_packages: list[str], - aur_packages: list[str], - user_packages: list[decman.UserPackage], - ignored_packages: list[str], - systemd_units: list[str], - systemd_user_units: dict[str, list[str]], + pacman_packages: set[str], + aur_packages: set[str], + user_packages: set[decman.UserPackage], + ignored_packages: set[str], + systemd_units: set[str], + systemd_user_units: dict[str, set[str]], files: dict[str, decman.File], directories: dict[str, decman.Directory], - modules: list[decman.Module], + modules: set[decman.Module], ): self.pacman_packages = pacman_packages self.aur_packages = aur_packages @@ -316,7 +349,7 @@ class Source: for target, directory in dirs.items(): try: print_debug(f"Installing directory to {target}.") - directory.copy_to(target, variables) + created_files.extend(directory.copy_to(target, variables)) except OSError as e: print_error(f"{e}") raise err.UserFacingError( @@ -332,6 +365,32 @@ class Source: return created_files + def all_file_targets(self) -> list[str]: + """ + Returns all file targets combined. + """ + all_files = [] + all_files.extend(self.files.keys()) + + for module in self.modules: + if module.enabled: + all_files.extend(module.files().keys()) + + return all_files + + def all_directory_targets(self) -> list[str]: + """ + Returns all directory targets combined. + """ + all_dirs = [] + all_dirs.extend(self.directories.keys()) + + for module in self.modules: + if module.enabled: + all_dirs.extend(module.directories().keys()) + + return all_dirs + def files_to_remove(self, store: Store, created_files: list[str]) -> list[str]: """ diff --git a/src/decman/lib/aur.py b/src/decman/lib/aur.py index 3d23387..23d687f 100644 --- a/src/decman/lib/aur.py +++ b/src/decman/lib/aur.py @@ -550,12 +550,12 @@ class ForeignPackageManager: def upgrade(self, upgrade_devel: bool = False, force: bool = False, - ignored_pkgs: typing.Optional[list[str]] = None): + ignored_pkgs: typing.Optional[set[str]] = None): """ Upgrades all foreign packages. """ if ignored_pkgs is None: - ignored_pkgs = [] + ignored_pkgs = set() l.print_summary("Determining packages to upgrade.") @@ -609,30 +609,17 @@ class ForeignPackageManager: resolved_dependencies = self.resolve_dependencies( foreign_pkgs, foreign_dep_pkgs) - l.print_summary( - "The following foreign packages will be installed explicitly:") - l.print_continuation("") - l.print_continuation( - f"\t{' '.join(resolved_dependencies.foreign_pkgs)}") - l.print_continuation("") + l.print_list_summary( + "The following foreign packages will be installed explicitly:", + list(resolved_dependencies.foreign_pkgs)) - if resolved_dependencies.foreign_dep_pkgs: - l.print_summary( - "The following foreign packages will be installed as dependencies:" - ) - l.print_continuation("") - l.print_continuation( - f"\t{' '.join(resolved_dependencies.foreign_dep_pkgs)}") - l.print_continuation("") + l.print_list_summary( + "The following foreign packages will be installed as dependencies:", + list(resolved_dependencies.foreign_dep_pkgs)) - if resolved_dependencies.foreign_build_dep_pkgs: - l.print_summary( - "The following foreign packages will be built in order to install other packages. They will not be installed:" - ) - l.print_continuation("") - l.print_continuation( - f"\t{' '.join(resolved_dependencies.foreign_build_dep_pkgs)}") - l.print_continuation("") + l.print_list_summary( + "The following foreign packages will be built in order to install other packages. They will not be installed:", + list(resolved_dependencies.foreign_build_dep_pkgs)) if not l.prompt_confirm("Proceed?", default=True): raise err.UserFacingError("Installing aborted.")