PipxInterface/Pipx.py

# -*- coding: utf-8 -*-

# Copyright (c) 2024 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the pipx GUI logic.
"""

import contextlib
import json
import os
import pathlib
import sysconfig

from PyQt6.QtCore import QObject, QProcess

from eric7 import Preferences
from eric7.EricWidgets import EricMessageBox
from eric7.SystemUtilities import OSUtilities

from .PipxExecDialog import PipxExecDialog


class Pipx(QObject):
    """
    Class implementing the pip GUI logic.
    """

    def __init__(self, parent=None):
        """
        Constructor

        @param parent reference to the user interface object
        @type QObject
        """
        super().__init__(parent)

        self.__ui = parent

    ############################################################################
    ## Utility methods
    ############################################################################

    def getPipxVersion(self):
        """
        Public method to get the version of the installed pipx package.

        @return string containing the pipx version number
        @rtype str
        """
        from pipx.version import version  # noqa: I102

        return version

    def getPipxVersionTuple(self):
        """
        Public method to get the version tuple of the installed pipx package.

        @return tuple containing the elements of the pipx version number
        @rtype tuple of (int, int, int)
        """
        from pipx.version import version_tuple  # noqa: I102

        return version_tuple

    def getPipxPaths(self):
        """
        Public method to get the paths used by pipx.

        @return dictionary containing the various pipx paths. The keys are
            'venvsPath', 'appsPath' and 'manPath'.
        @rtype dict[str, Path]
        """
        from pipx.paths import ctx  # noqa: I102

        return {
            "venvsPath": ctx.venvs,
            "appsPath": ctx.bin_dir,
            "manPath": ctx.man_dir,
            "pythonPath": ctx.standalone_python_cachedir,
        }

    def getPipxStrPaths(self):
        """
        Public method to get the paths used by pipx.

        @return dictionary containing the various pipx paths. The keys are
            'venvsPath', 'appsPath' and 'manPath'.
        @rtype dict[str, str]
        """
        from pipx.paths import ctx  # noqa: I102

        return {
            "venvsPath": str(ctx.venvs),
            "appsPath": str(ctx.bin_dir),
            "manPath": str(ctx.man_dir),
            "pythonPath": str(ctx.standalone_python_cachedir),
        }

    def __getPipxExecutable(self):
        """
        Private method to get the path name of the pipx executable.

        @return path name of the pipx executable
        @rtype str
        """
        binDir = sysconfig.get_path("scripts")
        pipx = os.path.join(binDir, "pipx")
        if OSUtilities.isWindowsPlatform():
            pipx += ".exe"

        return pipx

    def runPipxProcess(self, args):
        """
        Public method to execute pipx with the given arguments.

        @param args list of command line arguments for pipx
        @type list of str
        @return tuple containing a flag indicating success and the output
            of the process
        @rtype tuple of (bool, str)
        """
        ioEncoding = Preferences.getSystem("IOEncoding")

        process = QProcess()
        process.start(self.__getPipxExecutable(), args)
        procStarted = process.waitForStarted()
        if procStarted:
            finished = process.waitForFinished(30000)
            if finished:
                if process.exitCode() == 0:
                    output = str(process.readAllStandardOutput(), ioEncoding, "replace")
                    return True, output
                else:
                    error = str(process.readAllStandardError(), ioEncoding, "replace")
                    msg = self.tr("<p>Message:{0}</p>").format(error) if error else ""
                    return (
                        False,
                        self.tr("<p>pipx exited with an error ({0}).</p>{1}").format(
                            process.exitCode(), msg
                        ),
                    )
            else:
                process.terminate()
                process.waitForFinished(2000)
                process.kill()
                process.waitForFinished(3000)
                return False, self.tr("pipx did not finish within 30 seconds.")

        return False, self.tr("pipx could not be started.")

    def __metadataDecoderHook(self, jsonDict):
        """
        Private method to allow the JSON decoding of Path objects of a spec metadata
        file as created by 'pipx list --json'.

        @param jsonDict JSON dictionary to be decoded
        @type dict
        @return decoded Path object or the dictionary unaltered
        @rtype dict or pathlib.Path
        """
        if jsonDict.get("__type__") == "Path" and "__Path__" in jsonDict:
            return pathlib.Path(jsonDict["__Path__"])
        return jsonDict

    ############################################################################
    ## pipx interpreter list function (modified from original to work here)
    ############################################################################

    def getPipxInterpretersList(self):
        """
        Public method returning a list all standalone interpreters.

        @return dictionary containing data of standalone interpreters
        @rtype dict
        """
        from pipx.commands.interpreter import (  # noqa: I102
            get_installed_standalone_interpreters,
            get_interpreter_users,
            get_venvs_using_standalone_interpreter,
        )
        from pipx.paths import ctx  # noqa: I102
        from pipx.venv import VenvContainer  # noqa: I102

        interpreters = get_installed_standalone_interpreters()
        venvs = get_venvs_using_standalone_interpreter(VenvContainer(ctx.venvs))

        interpretersDict = {
            "path": str(ctx.standalone_python_cachedir),
            "interpreters": {},
        }
        for interpreter in interpreters:
            usedBy = get_interpreter_users(interpreter, venvs)
            interpretersDict["interpreters"][interpreter.name] = {
                "used": bool(usedBy),
                "used_by": [
                    (p.main_package.package, p.main_package.package_version)
                    for p in usedBy
                ],
            }
        return interpretersDict

    ############################################################################
    ## Command methods
    ############################################################################

    def getInstalledPackages(self):
        """
        Public method to get the installed packages.

        @return list of dictionaries containing the installed packages and apps
        @rtype list of dict[str, str | list]
        """
        from pipx.paths import ctx  # noqa: I102

        packages = []

        ok, output = self.runPipxProcess(["list", "--json"])
        if ok and output:
            with contextlib.suppress(json.JSONDecodeError):
                data = json.loads(output, object_hook=self.__metadataDecoderHook)
                for venvName in data["venvs"]:
                    metadata = data["venvs"][venvName]["metadata"]
                    package = {
                        "name": venvName,
                        "version": metadata["main_package"]["package_version"],
                        "apps": [],
                        "python": metadata["python_version"],
                        "is_standalone": (
                            str(metadata["source_interpreter"]).startswith(
                                str(ctx.standalone_python_cachedir.resolve())
                            )
                            if metadata["source_interpreter"]
                            else False
                        ),
                    }
                    for appPath in metadata["main_package"]["app_paths"]:
                        package["apps"].append((appPath.name, str(appPath)))
                    packages.append(package)

        return packages

    def installPackages(
        self,
        packages,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        forceVenvModification=False,
        systemSitePackages=False,
    ):
        """
        Public method to install a list of packages with the given options.

        @param packages list of packages to install
        @type list of str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param forceVenvModification flag indicating to allow modification of already
            existing virtual environments (defaults to False)
        @type bool (optional)
        @param systemSitePackages flag indicating to give access to the system
            site-packages directory (defaults to False)
        @type bool (optional)
        """
        args = ["install"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if forceVenvModification:
            args.append("--force")
        if systemSitePackages:
            args.append("--system-site-packages")
        args += packages
        dia = PipxExecDialog(self.tr("Install Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def installAllPackages(
        self,
        specFile,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        forceVenvModification=False,
        systemSitePackages=False,
    ):
        """
        Public method to install all packages define by a given spec metadata file
        with given options.

        @param specFile path of the spec metadata file
        @type str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param forceVenvModification flag indicating to allow modification of already
            existing virtual environments (defaults to False)
        @type bool (optional)
        @param systemSitePackages flag indicating to give access to the system
            site-packages directory (defaults to False)
        @type bool (optional)
        """
        args = ["install-all"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if forceVenvModification:
            args.append("--force")
        if systemSitePackages:
            args.append("--system-site-packages")
        args.append(specFile)
        dia = PipxExecDialog(self.tr("Install All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def createSpecMetadataFile(self, specFile):
        """
        Public method to create a spec metadata file.

        @param specFile path of the spec metadata file
        @type str
        @return tuple containing a flag indicating success and an error message in case
            of failure
        @rtype tuple of (bool, str)
        """
        ok, output = self.runPipxProcess(["list", "--json"])
        if ok:
            try:
                with open(specFile, "w") as f:
                    f.write(output)
                    return True, ""
            except IOError as err:
                return False, str(err)
        else:
            return False, output

    def reinstallPackage(
        self,
        package,
        interpreterVersion="",
        fetchMissingInterpreter=False,
    ):
        """
        Public method to reinstall the given package with given options.

        @param package name of the package to reinstall
        @type str
        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        """
        args = ["reinstall"]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        args.append(package)
        dia = PipxExecDialog(self.tr("Re-Install Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def reinstallAllPackages(
        self,
        interpreterVersion="",
        fetchMissingInterpreter=False,
        skipPackages=None,
    ):
        """
        Public method to reinstall all packages with given options.

        @param interpreterVersion version of the Python interpreter (defaults to "")
        @type str (optional)
        @param fetchMissingInterpreter flag indicating to fetch a standalone Python
            build from GitHub if the specified Python version is not found locally
            on the system (defaults to False)
        @type bool (optional)
        @param skipPackages list of packages to be skipped by the 'reinstall-all'
            command (defaults to None)
        @type list of str (optional)
        """
        args = ["reinstall-all"]
        if interpreterVersion:
            args += ["--python", interpreterVersion]
            if fetchMissingInterpreter:
                args.append("--fetch-missing-python")
        if skipPackages:
            args += ["--skip"] + skipPackages
        dia = PipxExecDialog(self.tr("Re-Install All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def uninstallPackage(self, package):
        """
        Public method to uninstall the given package.

        @param package name of the package to be uninstalled
        @type str
        """
        args = ["uninstall", package]
        dia = PipxExecDialog(self.tr("Uninstall Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def uninstallAllPackages(self):
        """
        Public method to uninstall all pipx managed packages.
        """
        args = ["uninstall-all"]
        dia = PipxExecDialog(self.tr("Uninstall All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def checkPackageOutdated(self, package):
        """
        Public method to check, if a given package is outdated.

        @param package name of the package
        @type str
        @return latest version in case the package is outdated and None otherwise
        @rtype str or None
        """
        args = ["runpip", package, "list", "--outdated", "--format", "json"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        ok, output = self.runPipxProcess(args)
        if not ok:
            EricMessageBox.information(
                None,
                self.tr("Check Outdated Package"),
                self.tr(
                    "<p>The status of package <b>{0}</b> could not be determined.</p>"
                    "<p>Reason: {1}</p>"
                ).format(package, output),
            )
            return None

        outdatedList = json.loads(output)
        # check if the main package is in the list
        for outdatedPackage in outdatedList:
            if outdatedPackage["name"] == package:
                return outdatedPackage["latest_version"]

        return None

    def upgradePackage(self, package):
        """
        Public method to upgrade the given package.

        @param package name of the package
        @type str
        """
        args = ["upgrade"]
        if Preferences.getPip("PipSearchIndex"):
            indexUrl = Preferences.getPip("PipSearchIndex") + "/simple"
            args += ["--index-url", indexUrl]
        args.append(package)
        dia = PipxExecDialog(self.tr("Upgrade Package"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeAllPackages(self):
        """
        Public method to upgrade all package.
        """
        args = ["upgrade-all"]
        dia = PipxExecDialog(self.tr("Upgrade All Packages"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeSharedLibraries(self):
        """
        Public method to upgrade shared libraries.
        """
        args = ["upgrade-shared"]
        dia = PipxExecDialog(self.tr("Upgrade Shared Libraries"))
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def upgradeInterpreters(self, dialogParent=None):
        """
        Public method to upgrade the installed interpreters to the latest available
        micro/patch version.

        @param dialogParent parent widget of the execution dialog
        @type QWidget
        """
        args = ["interpreter", "upgrade"]
        dia = PipxExecDialog(self.tr("Upgrade Interpreters"), parent=dialogParent)
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()

    def pruneInterpreters(self, dialogParent=None):
        """
        Public method to prune unused interpreters.

        @param dialogParent parent widget of the execution dialog
        @type QWidget
        """
        args = ["interpreter", "prune"]
        dia = PipxExecDialog(self.tr("Prune Unused Interpreters"), parent=dialogParent)
        res = dia.startProcess(self.__getPipxExecutable(), args)
        if res:
            dia.exec()