Skip to content

SOURCE CODE pipen.cli.plugins DOCS

"""List plugins"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any, Iterable, List, Tuple

from rich import print

from ._hooks import CLIPlugin
from ..defaults import (
    CLI_ENTRY_GROUP,
    SCHEDULER_ENTRY_GROUP,
    TEMPLATE_ENTRY_GROUP,
)
from ..utils import load_entrypoints

if TYPE_CHECKING:
    from argx import ArgumentParser
    from argparse import Namespace


COMMAND = "plugins"
GROUPS = [
    "pipen",
    SCHEDULER_ENTRY_GROUP,
    TEMPLATE_ENTRY_GROUP,
    CLI_ENTRY_GROUP,
]
GROUP_NAMES = {
    "pipen": "Pipen",
    SCHEDULER_ENTRY_GROUP: "Scheduler",
    TEMPLATE_ENTRY_GROUP: "Template",
    CLI_ENTRY_GROUP: "CLI",
}

__all__ = ("CliPluginsPlugin",)


def _get_plugins_by_group(group: str) -> Iterable[Tuple[str, Any]]:
    """Get plugins from entry points by group name

    Args:
        group: The name of the group

    Returns:
        A list of tuples with the plugin name and the plugin itself
    """
    for name, obj in load_entrypoints(group):
        yield name, obj


def _list_group_plugins(
    group: str,
    plugins: List[Tuple[str, Any]],
) -> None:
    """List plugins in a single group

    Args:
        group: The group of the plugins
        plugins: A list of tuples with name and plugin
    """
    print("")
    print(f"[bold][u]{GROUP_NAMES[group]} plugins:[/u][/bold]")
    namelen = max(len(name) for name, _ in plugins) if plugins else 0
    for name, plugin in plugins:
        try:
            ver = plugin.version
        except AttributeError:
            try:
                ver = plugin.__version__
            except AttributeError:
                ver = "unknown"
        print(f"- {name.ljust(namelen)}: (version: {ver})")


def _list_plugins(plugins: List[Tuple[str, str, Any]]) -> None:
    """List plugins

    Args:
        plugins: A list of tuples with group, name and plugin
    """
    pipen_plugins = [
        (name, plugin) for group, name, plugin in plugins if group == "pipen"
    ]
    sched_plugins = [
        (name, plugin)
        for group, name, plugin in plugins
        if group == SCHEDULER_ENTRY_GROUP
    ]
    tpl_plugins = [
        (name, plugin)
        for group, name, plugin in plugins
        if group == TEMPLATE_ENTRY_GROUP
    ]
    cli_plugins = [
        (name, plugin)
        for group, name, plugin in plugins
        if group == CLI_ENTRY_GROUP
    ]
    _list_group_plugins("pipen", pipen_plugins)
    _list_group_plugins(SCHEDULER_ENTRY_GROUP, sched_plugins)
    _list_group_plugins(TEMPLATE_ENTRY_GROUP, tpl_plugins)
    _list_group_plugins(CLI_ENTRY_GROUP, cli_plugins)


class CliPluginsPlugin(CLIPlugin):DOCS
    """List installed plugins"""

    name = "plugins"

    def __init__(
        self,
        parser: ArgumentParser,
        subparser: ArgumentParser,
    ) -> None:
        super().__init__(parser, subparser)
        subparser.add_argument(
            "-g",
            "--group",
            choices=GROUPS + ["all"],
            default="all",
            help="The name of the entry point group. Show all if not provided",
        )

    def exec_command(self, args: Namespace) -> None:DOCS
        """Execute the command"""
        from ..version import __version__
        print("Pipen version:", __version__)

        plugins: List[Tuple[str, str, Any]] = []

        if args.group and args.group != "all":
            for name, plugin in _get_plugins_by_group(args.group):
                plugins.append((args.group, name, plugin))

        else:  # args.name
            for group in GROUPS:
                for name, plugin in _get_plugins_by_group(group):
                    plugins.append((group, name, plugin))

        _list_plugins(plugins)