5
\$\begingroup\$

pacdef is my declarative package manager for Arch Linux. I have written it since I often found myself discovering a new tool at one of my PCs, using it there for a while, and then missing it when I'm at another PC. So I was looking for a consistent state of software on all my Linux machines.

It works like this: In my personal dotfiles repository, I have package files for different use cases, e.g. a file base, for all things that I need unconditionally (editor, kernel, python), a file for all programs that I need for work, etc..., and then a file for each machine (where I have stored, e.g., packages that provide drivers that I only need on one machine).

I can pacdef import a file to make pacdef aware of it, then install all packages from it via pacdef sync, remove unmanaged packages via pacdef clean, or review them via pacdef review.

More information can be found in the README.

Full code here: https://github.com/steven-omaha/pacdef

Current HEAD for purposes of this review: 67a9fdb

If you run Arch Linux or any of the derivatives, you can install it from the AUR: https://aur.archlinux.org/packages/pacdef

Main focus for the review should be the main.py file, which I have pasted below. Of course, all other files can be reviewed as well. Especially, I am interested in:

  • Is there too much overhead? Am I obscuring what's happening somewhere?
  • Are my abstractions appropriate?
  • I recently refactored the main class Pacdef and extracted a couple of functions, since it was doing too much. Would you consider it to be of appropriate scope and size?

Code smells I am aware of:

  • There are some tests (like 2 out of 50) I broke during refactoring and are ignored right now since I didn't know yet how to fix those. Those are deactivated. But the program works fine.

main.py

from __future__ import annotations

import io
import logging
import sys
from importlib.metadata import version
from os import environ
from typing import Callable, Optional

from .args import Arguments
from .aur_helper import AURHelper
from .cmd import run
from .config import Config
from .constants import EXIT_ERROR, EXIT_INTERRUPT, EXIT_SUCCESS, NOTHING_TO_DO, Action
from .db import DB
from .group import Group
from .path import file_exists
from .review import Reviewer
from .solver import (
    calc_packages_to_install,
    calc_unmanaged_packages,
    get_groups_matching_arguments,
    get_managed_packages,
)
from .user_input import get_user_confirmation


def main():
    """Run the main program."""
    _setup_logger()
    args = Arguments()
    config = Config()
    helper = AURHelper.from_config(config)
    db = DB()
    pacdef = Pacdef(args=args, config=config, aur_helper=helper, db=db)
    pacdef.run_action_from_arg()


def _setup_logger() -> None:
    """Set up the logger.

    When the log level is below WARNING (i.e. INFO or DEBUG), the line number of the logging statement is printed as
    well.
    """
    try:
        level_name: str = environ["LOGLEVEL"]
    except KeyError:
        level_name = "WARNING"

    level: int = logging.getLevelName(level_name.upper())
    if level < logging.WARNING:
        logging.basicConfig(format="%(levelname)s:%(lineno)d: %(message)s", level=level)
    else:
        logging.basicConfig(format="%(levelname)s: %(message)s", level=level)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(EXIT_INTERRUPT)


def _show_version() -> None:
    """Print version information to STDOUT.

    The value of `VERSION` is set during compile time by the PKGBUILD using `build()`.
    """
    print(f"pacdef, version: {version('pacdef')}")


class Pacdef:
    """Class representing the main routines of pacdef."""

    def __init__(
        self,
        args: Optional[Arguments] = None,
        config: Optional[Config] = None,
        aur_helper: Optional[AURHelper] = None,
        db: Optional[DB] = None,
        groups: Optional[list[Group]] = None,
    ):
        """Save the provided arguments as attributes, or use defaults when none are provided."""
        self._conf = config or Config()
        self._args = args or Arguments()
        self._aur_helper = aur_helper or AURHelper(self._conf.aur_helper)
        self._groups: list[Group] = groups or Group.read_groups_from_dir(
            self._conf.groups_path
        )
        self._db: DB = db or DB()
        self._sanity_check()

    @property
    def _action_map(self) -> dict[Action, Callable[[], None]]:
        """Return a dict matching all actions to their corresponding Pacdef methods."""
        return {
            Action.clean: self._remove_unmanaged_packages,
            Action.edit: self._edit_group_file,
            Action.groups: self._list_groups,
            Action.import_: self._import_groups,
            Action.new: self._new_group,
            Action.remove: self._remove_group,
            Action.review: self._review,
            Action.search: self._search_package,
            Action.show: self._show_group,
            Action.sync: self._install_packages_from_groups,
            Action.unmanaged: self._show_unmanaged_packages,
            Action.version: _show_version,
        }

    def _edit_group_file(self) -> None:
        logging.info("editing group files")
        groups = get_groups_matching_arguments(self._args, self._groups)
        if groups is None:
            logging.error("Could not find a suitable group")
            sys.exit(EXIT_ERROR)
        paths = [str(group.path) for group in groups]
        run([str(self._conf.editor), *paths], check=True)

    def _new_group(self) -> None:
        if self._args.groups is None:
            logging.error("Cannot create new group. No name supplied.")
            exit(EXIT_ERROR)

        # check if we can create all groups before we actually create them
        group_names = [g.name for g in self._groups]
        for group in self._args.groups:
            if group in group_names:
                logging.error(f"Cannot create new group '{group}', it already exists.")
                exit(EXIT_ERROR)

        for group in self._args.groups:
            Group.new_file(group, self._conf.groups_path)

        if self._args.edit_new:
            self._groups = Group.read_groups_from_dir(self._conf.groups_path)
            self._edit_group_file()

    def run_action_from_arg(self) -> None:
        """Get the function from the provided action arg, execute the function."""
        if self._args.action is not None:
            self._action_map[self._args.action]()

    def _review(self) -> None:
        unmanaged = calc_unmanaged_packages(
            get_managed_packages(self._groups),
            self._db.get_explicitly_installed_packages(),
        )

        Reviewer(self._groups, unmanaged, self._aur_helper).main()

    def _remove_unmanaged_packages(self) -> None:
        """Remove packages not managed by pacdef.

        Fetches unmanaged packages, then asks the user to confirm removing the packages. Then removes them using
        the AUR helper.
        """
        unmanaged_packages = calc_unmanaged_packages(
            get_managed_packages(self._groups),
            self._db.get_explicitly_installed_packages(),
        )
        if len(unmanaged_packages) == 0:
            print(NOTHING_TO_DO)
            sys.exit(EXIT_SUCCESS)
        print("Would remove the following packages and their dependencies:")
        for package in unmanaged_packages:
            print(f"  {package}")
        try:
            get_user_confirmation()
        except io.UnsupportedOperation:
            pass
        self._aur_helper.remove(unmanaged_packages)

    def _list_groups(self):
        """Print names of the imported groups to STDOUT."""
        groups_names = [group.name for group in self._groups]
        for name in groups_names:
            print(name)

    def _import_groups(self) -> None:
        if self._args.files is None:
            return
        for path in self._args.files:
            link_target = self._conf.groups_path / path.name
            if file_exists(link_target):
                logging.warning(f"{path.name} already exists, skipping")
            else:
                link_target.symlink_to(path.absolute())

    def _remove_group(self) -> None:
        """Remove the provided groups from the pacdef groups directory.

        More than one group can be provided. This method is atomic: If not all groups are found, none are removed.
        """
        found_groups = get_groups_matching_arguments(self._args, self._groups)
        if found_groups is None:
            logging.error("Could not find a suitable group")
            sys.exit(EXIT_ERROR)
        for group in found_groups:
            group.remove()

    def _search_package(self):
        """Show imported group which contains `_args.package`.

        The package name may be a regex. Only one package may be provided in the args.
        Exits with `EXIT_ERROR` if the package cannot be found.
        """
        if self._args.package is None:
            logging.error("no search string provided")
            sys.exit(EXIT_ERROR)

        matches = [
            (group, package)
            for group in self._groups
            for package in group
            if package.matches_regex(self._args.package)
        ]

        for group, package in matches:
            print(f"{group.name}: {package}")

        if matches:
            sys.exit(EXIT_SUCCESS)
        sys.exit(EXIT_ERROR)

    def _show_group(self) -> None:
        """Show all packages required by an imported group.

        More than one group may be provided, which prints the contents of all groups in order.
        """
        found_groups = get_groups_matching_arguments(self._args, self._groups)
        if found_groups is None:
            logging.error("Could not find a suitable group")
            sys.exit(EXIT_ERROR)
        for group in found_groups:
            print(group.content)

    def _install_packages_from_groups(self) -> None:
        """Install all packages from the imported package groups."""
        to_install = calc_packages_to_install(
            get_managed_packages(self._groups), self._db.get_all_installed_packages()
        )

        if len(to_install) == 0:
            print(NOTHING_TO_DO)
            sys.exit(EXIT_SUCCESS)
        print("Would install the following packages:")
        for package in to_install:
            print(f"  {package}")
        try:
            get_user_confirmation()
        except io.UnsupportedOperation:
            pass
        self._aur_helper.install(to_install)

    def _show_unmanaged_packages(self) -> None:
        """Print unmanaged packages to STDOUT."""
        unmanaged_packages = calc_unmanaged_packages(
            get_managed_packages(self._groups),
            self._db.get_explicitly_installed_packages(),
        )
        for package in unmanaged_packages:
            print(package)

    def _sanity_check(self) -> None:
        if self._conf.warn_symlinks:
            for group in self._groups:
                if not group.path.is_symlink():
                    logging.warning(f"group '{group.name}' is not a symlink")


Reply to @Reinderien answer

Pacdef._action_map must not be cached. It is evaluated exactly once during the runtime of the program. Caching would add overhead without benefits. It also cannot be a class variable AFAIK. If I were to write it like this:

class Pacdef:
    _action_map = {
        Action.clean: Pacdef._remove_unmanaged_packages,
        ...
    }

the interpreter cannot figure this out, since the Pacdef class does not exist yet. The only way I found that seemed somewhat idiomatic is making it a property.

NOTHING_TO_DO is used 3 times throughout the entire code. I would prefer to keep it that way: If I decide to change the message, I need to only change it in one place.

I have pushed some commits that partially implement Reinderien's answer. See here: https://github.com/steven-omaha/pacdef/compare/67a9fdb..6e8d3966bc

\$\endgroup\$

1 Answer 1

3
\$\begingroup\$

Minor:

print(f"pacdef, version: {version('pacdef')}")

doesn't need formatting, because the default separator for print is a space, thus

print('pacdef, version:', version('pacdef'))

_action_map, as a property, should either be cached or set as a class (i.e. static) variable. In the latter case, the initialisation could be done directly in class scope (not in a method), and the dictionary values would be to non-bound method references instead of bound method references.

_new_group calling exit is C-like. Though exit does technically throw an exception, the more idiomatic thing to do here is to throw an exception of your own, for it to be more practically caught by logic above if and when that's called for. Your main already has an except block converting exceptions to exit codes; you can just extend that.

group_names should be a set comprehension {} instead of a list comprehension [].

NOTHING_TO_DO being saved to a constant is strange. That might be called for if you're doing i18n, but you aren't, so just... write the string literal?

This block:

    if self._args.files is None:
        return
    for path in self._args.files:

can use a tuple to coalesce and drop the if, as in

for path in self._args.files or ():

Better yet, change self._args.files to default to that empty tuple instead of None.

\$\endgroup\$
1
  • 1
    \$\begingroup\$ Thank you for the review! Not all your comments apply to the code. I will add the corresponding information in my post later today. Still, insightful review! \$\endgroup\$
    – sarema
    Sep 26, 2022 at 11:46

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.