Source code for megu.plugin.manage

# -*- encoding: utf-8 -*-
# Copyright (c) 2021 Stephen Bunn <>
# GPLv3 License <>

"""Contains logic to install plugins into a directory."""

import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Optional

from ..config import instance as config
from ..helpers import temporary_directory
from ..log import instance as log
from .discover import discover_plugins

DIST_INFO_PATTERN = r"^(?P<package>.*)-.*\.dist-info$"

def _get_package_name(dirpath: Path) -> Optional[str]:
    """Attempt to extract an installed package's name from it's installed directory.

    .. important::
        This function assumes that the desired package is the ONLY package installed
        in the given directory. Otherwise, this function will return the first
        discovered package ``.dist-info`` and it's name. Please ensure that you are
        using this function in the appropriate context.

        dirpath (~pathlib.Path):
            The directory path where the desired package is the only package installed.

            If the given ``dirpath`` does not exist.

            The name of the package from ``dist-info`` if available and parseable.

    if not dirpath.is_dir():
        raise NotADirectoryError(f"No such directory {dirpath!s} exists")

    for path in dirpath.iterdir():
        if path.is_file():

        matches = re.match(DIST_INFO_PATTERN,
        if not matches:

        plugin_name = matches.groupdict().get("package", None)
        if not plugin_name:

        return plugin_name

    return None

[docs]def remove_plugin(package: str, plugin_dirpath: Optional[Path] = None): """Remove the given package if it exists in the plugin directory. Args: package (str): The name of the package to remove. plugin_dirpath (~pathlib.Path, optional): The plugin directory to remove the package from. Defaults to :attr:`~megu.constants.PLUGIN_DIR`. Raises: NotADirectoryError: If the given package does not exist as a subdirectory within the given plugin directory. """ if plugin_dirpath is None: plugin_dirpath = config.plugin_dir package_dirpath = plugin_dirpath.joinpath(package) if not package_dirpath.is_dir(): raise NotADirectoryError(f"No such directory {package_dirpath!s} exists") log.debug(f"Removing the plugin package at {package_dirpath}") shutil.rmtree(package_dirpath)
[docs]def add_plugin( package: str, plugin_dirpath: Optional[Path] = None, silence_subprocess: bool = False, ) -> Path: """Install a plugin utilizing pip. .. important:: If your package is not installable via pip through any of the distribution methods that pip checks (pypi, git, local, etc.), installation of your plugin simply will not work. Args: package (str): The package identifier that pip should use to discover and install your plugin. plugin_dirpath (~pathlib.Path, optional): The directory the plugin should be installed to. Defaults to :attr:`~megu.constants.PLUGIN_DIR`. silence_subprocess (bool): If set to ``True``, will redirect output of subprocess calls to /dev/null. Defaults to ``False``. Returns: ~pathlib.Path: The directory the plugin was installed to. """ if plugin_dirpath is None: plugin_dirpath = config.plugin_dir out_handle = subprocess.DEVNULL if silence_subprocess else sys.stdout err_handle = subprocess.DEVNULL if silence_subprocess else sys.stderr with temporary_directory("plugin-install-") as temp_dirpath: call_args = [ sys.executable, "-m", "pip", "install", "--upgrade", package, "--target", temp_dirpath.as_posix(), "--progress-bar", "off", "--no-color", ] try:"Installing plugin from {package!s} via pip using {call_args!r}") subprocess.check_call( call_args, stdout=out_handle, # type: ignore stderr=err_handle, # type: ignore ) except subprocess.CalledProcessError: log.exception( f"Failed to install plugin from {package!s} via pip using {call_args!r}" ) raise package_name = _get_package_name(temp_dirpath) if package_name is None: raise ValueError( f"Failed to extract package name from package at {temp_dirpath}" ) installed = list(discover_plugins(temp_dirpath)) if len(installed) <= 0: raise ValueError(f"Package at {temp_dirpath} exposes no plugins") package_dirpath = plugin_dirpath.joinpath(package_name) if package_dirpath.is_dir(): raise IsADirectoryError( f"Plugin directory at {package_dirpath!s} already exists" ) log.debug( f"Moving installed package {package_name} at {temp_dirpath} to " f"{package_dirpath}" ) shutil.copytree(temp_dirpath, package_dirpath)"Installed plugin {package_name} to {package_dirpath}") return package_dirpath