shepherd-agent/shepherd/agent/plugin.py

import importlib
from pathlib import Path
import inspect
import logging
import sys
import pkgutil
import pkg_resources
from configspec import ConfigSpecification
from .. import base_plugins


log = logging.getLogger(__name__)


class PluginLoadError(Exception):
    pass


def plugin(cls):
    return cls


def plugin_function():
    def wrapped(method):
        return method
    return wrapped


def plugin_hook(method):
    return method


def plugin_attachment(hookname):
    def wrapped(method):
        return method
    return wrapped


class PluginInterface():
    @staticmethod
    def _load_interface(module, plugin_name) -> "PluginInterface":
        """
        Finds PluginInterface instance in a plugin and returns it.
        """
        def is_plugininterface(member):
            return isinstance(member, PluginInterface)

        interface_list = inspect.getmembers(module, is_plugininterface)
        if not interface_list:
            raise PluginLoadError("Imported shepherd plugins must contain an instance"
                                  " of PluginInterface")

        if len(interface_list) > 1:
            log.warning(F"Plugin module {module.__name__} has more"
                        F" than one PluginInterface instance.")

        _, interface = interface_list[0]
        interface._plugin_name = plugin_name
        return interface

    def _load_confspec(self, module):
        """
        If not already registered, looks for a ConfigSpecification instance in a plugin,
        using a blank one by default.
        """
        if self._confspec is not None:
            return

        def is_confspec(member):
            return isinstance(member, ConfigSpecification)

        confspec_list = inspect.getmembers(module, is_confspec)
        if not confspec_list:
            self._confspec = ConfigSpecification()

        if len(confspec_list) > 1:
            log.warning(F"Plugin {self._plugin_name} has more"
                        F" than one root ConfigSpecification instance.")

        self.register_confspec(confspec_list[0][1])

    def _load_pluginclass(self, module):
        pass

    def __init__(self):
        self._confspec = None
        self._loaded = False
        self._plugin_name = "<not yet loaded>"

    def _load_guard(self):
        if self._loaded:
            raise PluginLoadError("Cannot call interface register functions once"
                                  " plugin is loaded")

    def register_confspec(self, confspec):
        self._load_guard()
        if not isinstance(confspec, ConfigSpecification):
            raise PluginLoadError("confspec must be an instance of ConfigSpecification")
        self._confspec = confspec

    @property
    def confspec(self):
        return self._confspec


def discover_base_plugins():
    """
    Returns a list of base plugin names available to load. These are plugins included with
    shepherd-agent, in 'base_plugins'.
    """
    return [name for _, name, _ in pkgutil.iter_modules(base_plugins.__path__)]


def discover_custom_plugins(plugin_dir=None):
    """
    Returns a list of custom plugin names available to load. This includes all modules or packages
    within the supplied custom plugin directory.
    """
    if plugin_dir:
        if Path(plugin_dir).is_dir():
            return [name for _, name, _ in pkgutil.iter_modules([plugin_dir])]
        else:
            log.warning(F"Custom plugin directory {plugin_dir} does not exist")
    return []


def discover_installed_plugins():
    """
    Returns a list of installed plugin names available to load. These are packages that have used
    the 'shephed.plugin' entrypoint in their setup.py
    """
    return [entrypoint.name for entrypoint in pkg_resources.iter_entry_points('shepherd.plugins')]


loaded_plugins = {}


def load_plugin(plugin_name, plugin_dir=None):
    """
    Finds a Shepherd plugin, loads it, and returns the resulting PluginInterface object.

    Will check 3 sources, in order:
        1. Built-in plugin modules/subpackages within ''shepherd.base_plugins''. Plugin
            module/package names match the plugin name.
        2. Modules/packages within the supplied ''plugin_dir'' path. Plugin module/package
            names match the plugin name.
        3. Any installed packages supplying the ''shepherd.plugin'' entrypoint.

    Args:
        plugin_name: Name used to try and locate the plugin
        plugin_dir: Optional directory path to be used for custom plugins

    Returns: The PluginInterface for the loaded plugin
    """

    if plugin_name in loaded_plugins:
        return loaded_plugins[plugin_name]

    # Each of the 3 plugin sources have different import mechanisms. Discovery is broken out to
    # allow them to be listed. Using a try/except block wouldn't be able to tell the difference
    # between a plugin not being found or //it's// imports not loading correctly.
    mod = None
    if plugin_name in discover_base_plugins():
        mod = importlib.import_module(base_plugins.__name__+'.'+plugin_name)
        log.info(F"Loading base plugin {plugin_name}")

    elif plugin_name in discover_custom_plugins(plugin_dir):
        saved_syspath = sys.path
        try:
            sys.path = [str(plugin_dir)]
            mod = importlib.import_module(plugin_name)
        finally:
            sys.path = saved_syspath
        modpath = getattr(mod, "__path__", [mod.__file__])[0]
        log.info(F"Loading custom plugin {plugin_name} from {modpath}")

    elif plugin_name in discover_installed_plugins():
        mod = pkg_resources.iter_entry_points('shepherd.plugins', plugin_name)[0].load()
        log.info(F"Loading installed plugin {plugin_name} from {mod.__name__}")

    if not mod:
        raise PluginLoadError("Could not find plugin "+plugin_name)

    interface = PluginInterface._load_interface(mod, plugin_name)
    interface._load_confspec(mod)

    # TODO Populate plugin interface

    loaded_plugins[plugin_name] = interface
    return interface


def init_plugins(plugin_configs, core_config):
    pass