Source code for _repobee.plugin

"""Plugin system module.

Module containing plugin system utility functions and classes.

.. module:: plugin
    :synopsis: PLugin system utility functions and classes.

.. moduleauthor:: Simon Larsén
import collections
import contextlib
import shutil
import tempfile
import pkgutil
import pathlib
import importlib
import os
from types import ModuleType
from typing import (

import _repobee
import _repobee.ext.defaults
import _repobee.ext.dist
import _repobee.ext.core_commands
import _repobee.distinfo
from _repobee import exception
from _repobee import featflags

import repobee_plug as plug

def _plugin_qualname(plugin_name):
    return f"{_repobee._internal_package_name}.ext.{plugin_name}"

def _external_plugin_qualname(plugin_name):
    return f"{_repobee._external_package_name}_{plugin_name}.{plugin_name}"

def load_plugin_modules(
    plugin_names: Iterable[str],
    allow_qualified: bool = False,
    allow_filepath: bool = False,
) -> List[ModuleType]:
    """Load the given plugins. Plugins are loaded such that they are executed
    in the same order that they are specified in the plugin_names list.

    When loading a plugin, tries to import first from :py:mod:`_repobee.ext`,
    and then from ``repobee_<plugin>``. For example, if ``javac`` is listed as
    a plugin, the following imports will be attempted:

    .. code-block:: python

        # import nr 1
        from _repobee.ext import javac

        # import nr 2
        from repobee_javac import javac

        # import nr 3 (only if allow_qualified)
        import javac

        # import nr 4 (only if allow_filepath)
        # Dynamically import using the name as a filepath

        plugin_names: A list of plugin names.
        allow_qualified: Allow the plugin to be specified by a qualified name.
        allow_filepath: Allows the plugin to be specified as a filepath.
        a list of loaded modules.
    loaded_modules = []
    plug.log.debug("Loading plugins: " + ", ".join(plugin_names))

    for name in plugin_names:
        plug_mod = (
            or _try_load_module(_external_plugin_qualname(name))
            or (allow_qualified and _try_load_module(name))
            or (allow_filepath and _try_load_module_from_filepath(name))
        if not plug_mod:
            msg = "failed to load plugin module " + name
            raise exception.PluginLoadError(msg)
    return loaded_modules

def _try_load_module_from_filepath(path: str) -> Optional[ModuleType]:
    """Try to load a module from the specified filepath.

    Adapted from code by Sebastian Rittau (

        path: A path to a Python module.
        The module if loaded successfully, or None if there was no module at
        the path.
    package_name = plug.fileutils.hash_path(path)
    module_name = pathlib.Path(path).stem
    qualname = f"{package_name}.{module_name}"
    spec = importlib.util.spec_from_file_location(qualname, path)
    if not spec:
        return None

    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)  # type: ignore

    return mod

def _try_load_module(qualname: str) -> Optional[ModuleType]:
    """Try to load a module.

        qualname: Qualified name of the module.

        the module if loaded properly, None otherwise
        return importlib.import_module(qualname)
    except ModuleNotFoundError as exc:
        return None

def register_plugins(
    modules: List[ModuleType],
) -> List[Union[plug.Plugin, ModuleType]]:
    """Register the namespaces of the provided modules, and any plug.Plugin
    instances in them. Registers modules in reverse order as they are
    run in LIFO order.

        modules: A list of modules.
        A list of registered modules and and plugin class instances.
    assert all([isinstance(mod, ModuleType) for mod in modules])

    registered: List[Union[plug.Plugin, ModuleType]] = []
    for module in reversed(modules):  # reverse because plugins are run LIFO
        plugin_name = module.__name__.split(".")[-1]

        for plugin_class in get_plugin_classes_in_module(module):
            obj = plugin_class(plugin_name)

    return registered

def get_plugin_classes_in_module(
    module: ModuleType,
) -> Iterable[Type[plug.Plugin]]:
    """Get all plugin classes in a module.

        module: A module.
        All classes that subtype :py:class:`plug.Plugin` in the module.
    for value in module.__dict__.values():
        if (
            isinstance(value, type)
            and issubclass(value, plug.Plugin)
            and value != plug.Plugin
            yield value

[docs]def unregister_all_plugins() -> None: """Unregister all currently registered plugins.""" for p in plug.manager.get_plugins(): plug.manager.unregister(p)
[docs]def try_register_plugin( plugin_module: ModuleType, *plugin_classes: List[type] ) -> None: """Attempt to register a plugin module and then immediately unregister it. .. important:: This is a convenience method for sanity checking plugins, and should only be called in test suites. It's not for production use. This convenience method can be used to sanity check plugins by registering them with RepoBee. If they have incorrectly defined hooks, this will be discovered only when registering. As an example, assume that we have a plugin module with a single (useless) plugin class in it, like this: .. code-block:: python :caption: import repobee_plug as plug class Useless(plug.Plugin): \"\"\"This plugin does nothing!\"\"\" We want to make sure that both the ``useless`` module and the ``Useless`` plugin class are registered correctly, and for that we can write some simple code like this. .. code-block:: python :caption: Example test case to check registering import repobee # assuming that useless is defined in the external plugin # repobee_useless from repobee_useless import useless def test_register_useless_plugin(): repobee.try_register_plugin(useless, useless.Useless) Args: plugin_module: A plugin module. plugin_classes: If the plugin contains any plugin classes (i.e. classes that extend :py:class:`repobee_plug.Plugin`), then these must be provided here. Otherwise, this option should not be provided. Raises: :py:class:`repobee_plug.PlugError` if the module cannot be registered, or if the contained plugin classes does not match plugin_classes. """ expected_plugin_classes = set(plugin_classes or []) newly_registered = register_plugins([plugin_module]) for reg in newly_registered: plug.manager.unregister(reg) registered_modules = [ reg for reg in newly_registered if isinstance(reg, ModuleType) ] registered_classes = { cl.__class__ for cl in newly_registered if cl not in registered_modules } assert len(registered_modules) == 1, "Module was not registered" if expected_plugin_classes != registered_classes: raise plug.PlugError( f"Expected plugin classes {expected_plugin_classes}, " f"got {registered_classes}" )
def initialize_plugins( plugin_names: List[str], allow_qualified: bool = False, allow_filepath: bool = False, ) -> List[Union[plug.Plugin, ModuleType]]: """Load and register plugins. Args: plugin_names: An optional list of plugin names that overrides the configuration file's plugins. allow_qualified: Allows the plugin names to be qualified. allow_filepath: Allows the plugin to be specified as a filepath. Returns: A list of registered modules and classes. Raises: :py:class:`_repobee.exception.PluginLoadError` """ if not allow_filepath: _check_no_filepaths(plugin_names) if not allow_qualified: _check_no_qualified_names(plugin_names) registered_plugins = plug.manager.get_plugins() plug_modules = [ p for p in load_plugin_modules( plugin_names, allow_qualified, allow_filepath ) if p not in registered_plugins ] registered = register_plugins(plug_modules) return registered def _is_filepath(name: str) -> bool: return os.pathsep in name or os.path.exists(name) def _check_no_filepaths(names: List[str]): filepaths = [name for name in names if _is_filepath(name)] if filepaths: raise exception.PluginLoadError(f"Filepaths not allowed: {filepaths}") def _check_no_qualified_names(names: List[str]): qualified_names = [ name for name in names if "." in name and not _is_filepath(name) ] if qualified_names: raise exception.PluginLoadError( f"Qualified names not allowed: {qualified_names}" ) def resolve_plugin_version(plugin_module: ModuleType) -> Optional[str]: """Return the version of this plugin. Tries to resolve the version by first checking if the plugin module itself has a ``__version__`` attribute, and then the top level package. Args: plugin_module: A plugin module. Returns: The version if found, otherwise None. """ if hasattr(plugin_module, "__version__"): return getattr(plugin_module, "__version__") assert plugin_module.__package__ is not None pkg_name = plugin_module.__package__.split(".")[0] pkg_module = _try_load_module(pkg_name) return getattr(pkg_module, "__version__", None) def is_default_plugin(module: ModuleType) -> bool: """Check if the provided module is a default module. Args: module: A Python module. Returns: True iff the provided module is a default plugin. """ return module.__package__ == _repobee.ext.defaults.__name__ def initialize_default_plugins() -> None: """Initialize the default plugin modules.""" default_plugin_qualnames = get_qualified_module_names( _repobee.ext.defaults ) if featflags.is_feature_enabled( featflags.FeatureFlag.REPOBEE_CORE_COMMANDS_AS_PLUGINS ): core_command_plugin_names = get_qualified_module_names( _repobee.ext.core_commands ) default_plugin_qualnames.extend(core_command_plugin_names) initialize_plugins(default_plugin_qualnames, allow_qualified=True) def initialize_dist_plugins(force: bool = False) -> None: """Initialize the distribution plugin modules. Args: force: Initialize dist plugins even if RepoBee has not been properly installed. """ if not _repobee.distinfo.DIST_INSTALL and not force: raise exception.PluginLoadError( "Dist plugins can only be loaded with installed RepoBee" ) dist_plugin_qualnames = get_qualified_module_names(_repobee.ext.dist) initialize_plugins(dist_plugin_qualnames, allow_qualified=True) def get_qualified_module_names(pkg: ModuleType) -> List[str]: """Return a list of all python modules in the given package. Only considers the modules directly in this package, and not in subpackages. Args: pkg: The package to resolve modules in. Returns: All modules in the given package. """ return [f"{pkg.__name__}.{name}" for name in get_module_names(pkg)] def get_module_names(pkg: ModuleType) -> List[str]: """Get the unqualified module names from the given package. Args: pkg: The package to resolve modules in. Returns: All modules in the given package. """ pkg_path = pkg.__path__ # type: ignore return [ name for file_finder, name, _ in pkgutil.iter_modules(pkg_path) # only include modules (i.e. files), not subpackages if (pathlib.Path(file_finder.path) / (name + ".py")).is_file() # type: ignore ] def execute_clone_tasks( repos: Iterable[plug.StudentRepo], api: plug.PlatformAPI, cwd: Optional[pathlib.Path] = None, ) -> Mapping[str, List[plug.Result]]: """Execute clone tasks, if there are any, and return the results. Args: repo_names: Names of the repositories to execute clone tasks on. api: An instance of the platform API. cwd: Directory in which to find the repos. Returns: A mapping from repo name to hook result. """ return execute_tasks(repos, plug.manager.hook.post_clone, api, cwd) def execute_setup_tasks( repos: Iterable[plug.TemplateRepo], api: plug.PlatformAPI, cwd: Optional[pathlib.Path] = None, ) -> Mapping[str, List[plug.Result]]: """Execute setup tasks, if there are any, and return the results. Args: repos: Template repos. api: An instance of the platform API. cwd: Directory in which to find the repos. Returns: A mapping from repo name to hook result. """ return execute_tasks( repos, plug.manager.hook.pre_setup, api, cwd, copy_repos=False ) def execute_tasks( repos: Iterable[Union[plug.StudentRepo, plug.TemplateRepo]], hook_function: Callable[ [pathlib.Path, plug.PlatformAPI], Optional[plug.Result] ], api: plug.PlatformAPI, cwd: Optional[pathlib.Path], copy_repos: bool = True, extra_kwargs: Optional[Mapping[Any, Any]] = None, ) -> Mapping[str, List[plug.Result]]: """Execute plugin tasks on the provided repos.""" cwd = cwd or pathlib.Path(".") with tempfile.TemporaryDirectory() as tmpdir: copies_root = pathlib.Path(tmpdir)"Executing tasks ...") all_results: DefaultDict[ str, List[plug.Result] ] = collections.defaultdict(list) for repo in ( _copy_repos(repos, basedir=copies_root) if copy_repos else repos ):"Processing {}") valid_results: List[plug.Result] = [result for result in hook_function(repo=repo, api=api, **(extra_kwargs or {})) if result] # type: ignore all_results[].extend(valid_results) return all_results def _copy_repos( repos: Iterable[Union[plug.StudentRepo, plug.TemplateRepo]], basedir: pathlib.Path, ) -> Iterable[Union[plug.StudentRepo, plug.TemplateRepo]]: for repo in repos: copy_path = basedir / plug.fileutils.hash_path(repo.path) shutil.copytree(repo.path, copy_path) yield repo.with_path(copy_path) @contextlib.contextmanager def _convert_task_exceptions(task): """Catch task exceptions and re-raise or convert into something more appropriate for the user. Only plug.PlugErrors will be let through without modification. """ try: yield except plug.PlugError as exc: raise plug.PlugError( f"A task from the module '{task.act.__module__}' crashed: {str(exc)}" ) except Exception as exc: raise plug.PlugError( f"A task from the module '{task.act.__module__}' crashed unexpectedly. " "This is a bug, please report it to the plugin " "author." ) from exc def _handle_deprecation(): """Emit warnings if any deprecated hooks are used.""" deprecated_hooks = plug.deprecated_hooks() deprecated_hook_names = deprecated_hooks.keys() for p in plug.manager.get_plugins(): for member in dir(p): if member in deprecated_hook_names: deprecation = deprecated_hooks[member] msg = ( f"A plugin from the module '{p.__module__ if '__module__' in dir(p) else p.__name__,}' is using the " f"deprecated '{member}' hook, which will stop being supported as " f"of RepoBee {deprecation.remove_by_version}." ) if deprecation.replacement: msg += ( f" '{deprecation.replacement}' should be used instead." ) plug.log.warning(msg)