Source code for ska_sdp_batchlet.plugins.utils

import importlib
from typing import TypeVar

__all__ = ("load_plugin", "instantiate_plugins")


BasePlugin = TypeVar("BasePlugin")




def load_plugin(
    plugin_base_class: type[BasePlugin], name: str, path: str = None, **_
) -> type[BasePlugin]:
    """
    Validates and returns a plugin class.

    Parameters
    ----------
    plugin_base_class
        The base class type that the plugin must inherit from.
    name
        The name of the plugin class to validate.
    path
        The module path of the plugin. If None, it will fallback
        to ``plugin_base_class.DEFAULT_PACKAGE_PATH``.

    Returns
    -------
        The validated plugin class.

    Raises
    ------
    ImportError
        If the specified class is not found in the module.
    TypeError
        If the class is not a subclass of `plugin_base_class`.
    """
    cls = None

    if path:
        spec = importlib.util.spec_from_file_location(
            f"__temp_plugin_module_{path.split('/')[-1].split('.py')[0]}", path
        )
        if spec is None:
            raise ImportError(f"Module not found at path '{path}'")
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
    else:
        module_path = plugin_base_class.DEFAULT_PACKAGE_PATH
        module = importlib.import_module(module_path)

    try:
        cls = getattr(module, name)
    except AttributeError as exc:
        raise ImportError(
            f"Class '{name}' not found in module '{module.__name__}'"
        ) from exc

    if not issubclass(cls, plugin_base_class):
        raise TypeError(
            f"{cls.__name__} must be a subclass of "
            f"{plugin_base_class.__name__}"
        )

    return cls





def instantiate_plugins(
    plugin_base_class: type[BasePlugin], plugin_configs: list[dict]
) -> list[BasePlugin]:
    """
    Instantiate plugins based on the configuration.

    Parameters
    ----------
    plugin_base_class
        The base class type that the plugins must inherit from.
    plugin_configs
        List of plugin configurations. Each configuration is a dictionary with
        following keys:

        - name: Name of the plugin class to load
        - path: (optional) Path to the python module which contains the
                plugin class
        - kwargs: (optional) Extra key-word argumments passed to the
                plugin class while instantiating.

    Returns
    -------
        List of fully initialized plugin objects

    Raises
    ------
    ImportError
        If the specified class is not found in the module.
    TypeError
        If the class is not a subclass of `plugin_base_class`.
    """
    return [
        load_plugin(plugin_base_class, **plugin_config)(
            **plugin_config.get("kwargs", {})
        )
        for plugin_config in plugin_configs
    ]