shepherd-agent/shepherd/agent/plugin.py

import importlib
from pathlib import Path
import inspect
import logging
import sys
import pkgutil
import pkg_resources
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():
    def __init__(self):
        self._confspec = None

    @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__)
        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)

    # Scan imported module for PluginInterface instance
    def is_plugininterface(member):
        return isinstance(member, PluginInterface)

    interface_list = inspect.getmembers(mod, 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 {mod.__name__} has more"
                    F" than one PluginInterface instance.")

    _, interface = interface_list[0]

    # TODO Populate plugin interface

    loaded_plugins[plugin_name] = interface
    return interface


def init_plugins(plugin_configs, core_config):
    pass