Source code for _repobee.main

"""Main entrypoint for the repobee CLI application.

.. module:: main
    :synopsis: Main entrypoint for the repobee CLI application.

.. moduleauthor:: Simon Larsén
"""

import argparse
import contextlib
import dataclasses
import enum
import io
import logging
import os
import pathlib
import sys
from typing import List, Optional, Union, Mapping, Any, NoReturn
from types import ModuleType

import repobee_plug as plug

import _repobee.cli.dispatch
import _repobee.cli.parsing
import _repobee.cli.preparser
import _repobee.cli.mainparser
import _repobee.constants
import _repobee.config
from _repobee import plugin
from _repobee import exception
from _repobee.cli.preparser import separate_args
from _repobee import distinfo
from _repobee import disthelpers


_PRE_INIT_ERROR_MESSAGE = """exception was raised before pre-initialization was
complete. This is usually due to incorrect settings.
Try running the `verify-settings` command and see if
the problem can be resolved. If all fails, please open
an issue at https://github.com/repobee/repobee/issues/new
and supply the stack trace below.""".replace(
    "\n", " "
)


[docs] def run( cmd: List[str], config_file: Union[str, pathlib.Path] = "", plugins: Optional[List[Union[ModuleType, plug.Plugin]]] = None, workdir: Union[str, pathlib.Path] = ".", ) -> Mapping[str, List[plug.Result]]: """Run RepoBee with the provided options. This function is mostly intended to be used for testing plugins. .. important:: This function will always unregister all plugins after execution, including only plugins that may have been registered prior to running this function. Running this function is almost equivalent to running RepoBee from the CLI, with the following exceptions: 1. Preparser options must be passed as arguments to this function (i.e. cannot be given as part of ``cmd``). 2. There is no error handling at the top level, so exceptions are raised instead of just logged. As an example, the following CLI call: .. code-block:: bash $ repobee --plug ext.py --config-file config.ini config show Can be executed as follows: .. code-block:: python import ext from repobee import run run(["config", "show"], config_file="config.ini", plugins=[ext]) Args: cmd: The command to run. config_file: Path to the configuration file. plugins: A list of plugin modules and/or plugin classes. workdir: The working directory to run RepoBee in. Returns: A mapping (plugin_name -> plugin_results). """ conf = _to_config(pathlib.Path(config_file)) requested_workdir = pathlib.Path(str(workdir)).resolve(strict=True) with _in_workdir(requested_workdir), _unregister_plugins_on_exit(): _initialize_logging_and_plugins_for_run(plugins or []) parsed_args, api = _parse_args(cmd, conf) output_verbosity = _get_output_verbosity(parsed_args) with _set_output_verbosity(output_verbosity): return _repobee.cli.dispatch.dispatch_command( parsed_args, api, conf )
def _wrap_in_plugin_module(maybe_plugin: Any) -> ModuleType: if isinstance(maybe_plugin, type) and issubclass( maybe_plugin, plug.Plugin ): mod = ModuleType(maybe_plugin.__name__.lower()) mod.__package__ = f"__{maybe_plugin.__name__}" setattr(mod, maybe_plugin.__name__, maybe_plugin) return mod elif isinstance(maybe_plugin, ModuleType): return maybe_plugin else: raise TypeError(f"not plugin or module: {maybe_plugin}") def _initialize_logging_and_plugins_for_run(plugins: List[Any]): wrapped_plugins = list(map(_wrap_in_plugin_module, plugins or [])) _repobee.cli.parsing.setup_logging() _initialize_mandatory_plugins() plugin.register_plugins(wrapped_plugins) @contextlib.contextmanager def _unregister_plugins_on_exit(unregister: bool = True): try: yield finally: if unregister: plugin.unregister_all_plugins() def main( sys_args: List[str], unload_plugins: bool = True, workdir: pathlib.Path = pathlib.Path(".").resolve(), ): """Start the repobee CLI. Args: sys_args: Arguments from the command line. unload_plugins: If True, plugins are automatically unloaded just before the function returns. workdir: The working directory to operate in. """ with _main_error_handler(), _in_workdir( workdir ), _unregister_plugins_on_exit(unregister=unload_plugins): _run_cli(sys_args) @contextlib.contextmanager def _main_error_handler(): try: yield except plug.PlugError: plug.log.error("A plugin exited with an error") sys.exit(1) except Exception: plug.log.error( "RepoBee exited unexpectedly. " "Please visit the FAQ to try to resolve the problem: " "https://repobee.readthedocs.io/en/stable/faq.html" ) sys.exit(1) def _run_cli(sys_args: List[str]): _repobee.cli.parsing.setup_logging() args = sys_args[1:] # drop the name of the program with _pre_init_error_handler(): app_init = _run_preparser_and_init_application(args) output_verbosity = _get_output_verbosity(app_init.parsed_args) show_traceback = app_init.parsed_args.traceback with _set_output_verbosity(output_verbosity), _core_error_handler( show_traceback ): _repobee.cli.dispatch.dispatch_command( app_init.parsed_args, app_init.platform_api, app_init.config ) @dataclasses.dataclass class _ApplicationInitialization: parsed_args: argparse.Namespace # FIXME platform_api should be optional, but typing error in dispatch_command prevents this platform_api: plug.PlatformAPI config: plug.Config def _run_preparser_and_init_application( args: List[str], ) -> _ApplicationInitialization: preparser_args, app_args = separate_args(args) parsed_preparser_args = _repobee.cli.preparser.parse_args( preparser_args, default_config_file=_resolve_config_file(pathlib.Path(".").resolve()), ) # IMPORTANT: the mandatory plugins must be loaded before user-defined # plugins to ensure that the user-defined plugins override the defaults # in firstresult hooks _initialize_mandatory_plugins() if not parsed_preparser_args.no_plugins: _initialize_non_default_plugins(parsed_preparser_args.plug or []) conf = _to_config(parsed_preparser_args.config_file) parsed_args, api = _parse_args(app_args, conf) return _ApplicationInitialization(parsed_args, api, conf) @contextlib.contextmanager def _pre_init_error_handler(): try: yield except ( exception.ParseError, exception.PluginLoadError, exception.FileError, ) as exc: plug.echo(_PRE_INIT_ERROR_MESSAGE) plug.log.error(f"{exc.__class__.__name__}: {exc}") raise except Exception as exc: plug.echo(_PRE_INIT_ERROR_MESSAGE) _handle_unexpected_exception(exc, traceback=True) @contextlib.contextmanager def _core_error_handler(traceback: bool): try: yield except Exception as exc: _handle_unexpected_exception(exc, traceback=True) def _handle_unexpected_exception(exc: Exception, traceback: bool) -> NoReturn: plug.log.error(f"{exc.__class__.__name__}: {exc}") if traceback: plug.log.exception("Critical exception") raise exc def _to_config(config_file: pathlib.Path) -> plug.Config: if config_file.is_file(): _repobee.config.check_config_integrity(config_file) return plug.Config(config_file) def _resolve_config_file(path: pathlib.Path) -> pathlib.Path: local_config_path = path / _repobee.constants.LOCAL_CONFIG_NAME if local_config_path.is_file(): return local_config_path elif path.parent == path: # file system root return _repobee.constants.DEFAULT_CONFIG_FILE else: return _resolve_config_file(path.parent) def _initialize_mandatory_plugins(): plug.log.debug("Initializing default plugins") plugin.initialize_default_plugins() if distinfo.DIST_INSTALL: plug.log.debug("Initializing dist plugins") plugin.initialize_dist_plugins() def _initialize_non_default_plugins(plugin_names: List[str]) -> None: if distinfo.DIST_INSTALL: plug.log.debug("Initializing active plugins") plugin.initialize_plugins( disthelpers.get_active_plugins(), allow_filepath=True ) plug.log.debug("Initializing preparser-specified plugins") plugin.initialize_plugins(plugin_names, allow_filepath=True) def _parse_args(args: List[str], config: plug.Config): _repobee.config.execute_config_hooks(config) parsed_args, api = _repobee.cli.parsing.handle_args(args, config) plug.manager.hook.handle_processed_args(args=parsed_args) return parsed_args, api class _OutputVerbosity(enum.IntEnum): SILENCE_ERRORS = -3 SILENCE_WARNINGS = -2 SILENCE_STDOUT = -1 STANDARD = 0 INFO_LOGGING = 1 DEBUG_LOGGING = 2 @contextlib.contextmanager def _set_output_verbosity(verbosity: _OutputVerbosity): """Set the output verbosity, expecting `quietness` to be a non-negative integer. """ if verbosity == _OutputVerbosity.STANDARD: yield return elif verbosity > _OutputVerbosity.STANDARD: terminal_level = ( logging.INFO if verbosity == _OutputVerbosity.INFO_LOGGING else logging.DEBUG ) _repobee.cli.parsing.setup_logging(terminal_level=terminal_level) yield else: # verbosity <= SILENCE_STDOUT # silence stdout by redirecting to internal buffer with contextlib.redirect_stdout(io.StringIO()): if verbosity == _OutputVerbosity.SILENCE_WARNINGS: _repobee.cli.parsing.setup_logging( terminal_level=logging.ERROR ) elif verbosity == _OutputVerbosity.SILENCE_ERRORS: _repobee.cli.parsing.setup_logging( terminal_level=logging.CRITICAL ) yield def _get_output_verbosity(parsed_args: argparse.Namespace) -> _OutputVerbosity: return _OutputVerbosity( -getattr(parsed_args, "quiet", 0) or getattr(parsed_args, "verbose", 0) ) @contextlib.contextmanager def _in_workdir(workdir: pathlib.Path): cur_workdir = pathlib.Path(".").resolve() try: os.chdir(workdir) yield finally: os.chdir(cur_workdir) if __name__ == "__main__": main(sys.argv)