ahriman/src/ahriman/application/handlers/handler.py

#
# Copyright (c) 2021-2023 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import argparse
import logging

from multiprocessing import Pool

from ahriman.application.lock import Lock
from ahriman.core.configuration import Configuration
from ahriman.core.exceptions import ExitCode, MissingArchitectureError, MultipleArchitecturesError
from ahriman.core.log import Log
from ahriman.models.repository_paths import RepositoryPaths


class Handler:
    """
    base handler class for command callbacks

    Attributes:
        ALLOW_AUTO_ARCHITECTURE_RUN(bool): (class attribute) allow defining architecture from existing repositories
        ALLOW_MULTI_ARCHITECTURE_RUN(bool): (class attribute) allow running with multiple architectures

    Examples:
        Wrapper for all command line actions, though each derived class implements ``run`` method, it usually must not
        be called directly. The recommended way is to call ``execute`` class method, e.g.::

            >>> from ahriman.application.handlers import Add
            >>>
            >>> Add.execute(args)
    """

    ALLOW_AUTO_ARCHITECTURE_RUN = True
    ALLOW_MULTI_ARCHITECTURE_RUN = True

    @classmethod
    def architectures_extract(cls, args: argparse.Namespace) -> list[str]:
        """
        get known architectures

        Args:
            args(argparse.Namespace): command line args

        Returns:
            list[str]: list of architectures for which tree is created

        Raises:
            MissingArchitecture: if no architecture set and automatic detection is not allowed or failed
        """
        if not cls.ALLOW_AUTO_ARCHITECTURE_RUN and args.architecture is None:
            # for some parsers (e.g. config) we need to run with specific architecture
            # for those cases architecture must be set explicitly
            raise MissingArchitectureError(args.command)
        if args.architecture:  # architecture is specified explicitly
            return sorted(set(args.architecture))

        configuration = Configuration()
        configuration.load(args.configuration)
        # wtf???
        root = configuration.getpath("repository", "root")  # pylint: disable=assignment-from-no-return
        architectures = RepositoryPaths.known_architectures(root)

        if not architectures:  # well we did not find anything
            raise MissingArchitectureError(args.command)
        return sorted(architectures)

    @classmethod
    def call(cls, args: argparse.Namespace, architecture: str) -> bool:
        """
        additional function to wrap all calls for multiprocessing library

        Args:
            args(argparse.Namespace): command line args
            architecture(str): repository architecture

        Returns:
            bool: True on success, False otherwise
        """
        try:
            configuration = Configuration.from_path(args.configuration, architecture)
            log_handler = Log.handler(args.log_handler)
            Log.load(configuration, log_handler, quiet=args.quiet, report=args.report)
            with Lock(args, architecture, configuration):
                cls.run(args, architecture, configuration, report=args.report)
            return True
        except ExitCode:
            return False
        except Exception:
            # we are basically always want to print error to stderr instead of default logger
            logging.getLogger("stderr").exception("process exception")
            return False

    @classmethod
    def execute(cls, args: argparse.Namespace) -> int:
        """
        execute function for all aru

        Args:
            args(argparse.Namespace): command line args

        Returns:
            int: 0 on success, 1 otherwise

        Raises:
            MultipleArchitectures: if more than one architecture supplied and no multi architecture supported
        """
        architectures = cls.architectures_extract(args)

        # actually we do not have to spawn another process if it is single-process application, do we?
        if len(architectures) > 1:
            if not cls.ALLOW_MULTI_ARCHITECTURE_RUN:
                raise MultipleArchitecturesError(args.command)

            with Pool(len(architectures)) as pool:
                result = pool.starmap(
                    cls.call, [(args, architecture) for architecture in architectures])
        else:
            result = [cls.call(args, architectures.pop())]

        return 0 if all(result) else 1

    @classmethod
    def run(cls, args: argparse.Namespace, architecture: str, configuration: Configuration, *, report: bool) -> None:
        """
        callback for command line

        Args:
            args(argparse.Namespace): command line args
            architecture(str): repository architecture
            configuration(Configuration): configuration instance
            report(bool): force enable or disable reporting

        Raises:
            NotImplementedError: not implemented method
        """
        raise NotImplementedError

    @staticmethod
    def check_if_empty(enabled: bool, predicate: bool) -> None:
        """
        check condition and flag and raise ExitCode exception in case if it is enabled and condition match

        Args:
            enabled(bool): if False no check will be performed
            predicate(bool): indicates condition on which exception should be thrown

        Raises:
            ExitCode: if result is empty and check is enabled
        """
        if enabled and predicate:
            raise ExitCode()