Add CLI

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

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]

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"}
 ]

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_dir, _, src_files in os.walk("."):
-                for src_file in files:
+                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] = []

src/decman/__main__.py

@@ -0,0 +1,3 @@
+import decman.app
+
+decman.app.main()

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()
+    if not _is_root():
-
+        l.print_error("Not running as root. Please run decman as root.")
-    print(aur.get_package_info("zapzap"))
-
-    if not is_root():
-        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:
+def _set_up(store: l.Store):
-    """
+    parser = argparse.ArgumentParser(
-    Returns True if the process is running as root.
+        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()

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],
+        pacman_packages: set[str],
-        aur_packages: list[str],
+        aur_packages: set[str],
-        user_packages: list[decman.UserPackage],
+        user_packages: set[decman.UserPackage],
-        ignored_packages: list[str],
+        ignored_packages: set[str],
-        systemd_units: list[str],
+        systemd_units: set[str],
-        systemd_user_units: dict[str, list[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]:
         """

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(
+        l.print_list_summary(
-            "The following foreign packages will be installed explicitly:")
+            "The following foreign packages will be installed explicitly:",
-        l.print_continuation("")
+            list(resolved_dependencies.foreign_pkgs))
-        l.print_continuation(
-            f"\t{' '.join(resolved_dependencies.foreign_pkgs)}")
-        l.print_continuation("")
 
-        if resolved_dependencies.foreign_dep_pkgs:
+        l.print_list_summary(
-            l.print_summary(
+            "The following foreign packages will be installed as dependencies:",
-                "The following foreign packages will be installed as dependencies:"
+            list(resolved_dependencies.foreign_dep_pkgs))
-            )
-            l.print_continuation("")
-            l.print_continuation(
-                f"\t{' '.join(resolved_dependencies.foreign_dep_pkgs)}")
-            l.print_continuation("")
 
-        if resolved_dependencies.foreign_build_dep_pkgs:
+        l.print_list_summary(
-            l.print_summary(
+            "The following foreign packages will be built in order to install other packages. They will not be installed:",
-                "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))
-            )
-            l.print_continuation("")
-            l.print_continuation(
-                f"\t{' '.join(resolved_dependencies.foreign_build_dep_pkgs)}")
-            l.print_continuation("")
 
         if not l.prompt_confirm("Proceed?", default=True):
             raise err.UserFacingError("Installing aborted.")