Sun, 29 Dec 2024 14:56:04 +0100
Prepared a new release.
# -*- coding: utf-8 -*- # Copyright (c) 2024 - 2025 Detlev Offenbach <detlev@die-offenbachs.de> # """ Module implementing the pipx GUI logic. """ import contextlib import functools import json import os import pathlib import sysconfig from packaging.specifiers import InvalidSpecifier, SpecifierSet from PyQt6.QtCore import QObject, QProcess, pyqtSignal from eric7 import Preferences from eric7.EricWidgets import EricMessageBox from eric7.SystemUtilities import OSUtilities, PythonUtilities try: from eric7.EricCore.EricProcess import EricProcess except ImportError: # backward compatibility for eric-ide < 24.10 from .PipxProcess import PipxProcess as EricProcess from .PipxExecDialog import PipxExecDialog class Pipx(QObject): """ Class implementing the pipx interface. @signal outdatedPackage(package:str, latestVer:str, oudatedDeps:bool) emitted with the result of a check for outdated status of a package """ outdatedPackage = pyqtSignal(str, str, bool) def __init__(self, parent=None): """ Constructor @param parent reference to the user interface object @type QObject """ super().__init__(parent) self.__ui = parent self.__pipxProcesses = [] def shutdown(self): """ Public method to perform shutdown actions. """ for proc in self.__pipxProcesses: proc.kill() self.__pipxProcesses.clear() ############################################################################ ## 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 """ ok, output = self.__runPipxProcess(["--version"]) if ok: return output.strip() else: return "" 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): """ Private 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 def __runPipxAsyncProcess(self, args, callback=None, timeout=30000): """ Private method to execute pipx with the given arguments asynchronously. @param args list of command line arguments for pipx @type list of str @param callback reference to the function to be called a success flag and the process output or error message (defaults to None) @type function (optional) @param timeout timeout for the process in milliseconds (defaults to 30000) @type int (optional) @return reference to the generated process object @rtype QProcess """ process = EricProcess(timeout=timeout) process.finished.connect( functools.partial(self.__asyncProcessFinished, callback, process) ) process.errorOccurred.connect( functools.partial(self.__asyncProcessError, process) ) self.__pipxProcesses.append(process) process.start(self.__getPipxExecutable(), args) return process def __asyncProcessError(self, process, error): """ Private method to handle a process error signal. @param process reference to the process @type QProcess @param error error that occurred @type QProcess.ProcessError """ if error == QProcess.ProcessError.FailedToStart: with contextlib.suppress(ValueError): self.__pipxProcesses.remove(process) EricMessageBox.critical( None, self.tr("pipx Start Error"), self.tr("pipx could not be started.") ) else: EricMessageBox.critical( None, self.tr("pipx Runtime Error"), self.tr( "<p>The pipx process reported an error.</p><p>Error: {0}</p>" ).format(process.errorString()), ) def __asyncProcessFinished(self, callback, process, _exitCode, exitStatus): """ Private method to handle the process finished signal. @param callback reference to the function to be called a success flag and the process output or error message @type function @param process reference to the process @type QProcess @param _exitCode exit code of the process @type int @param exitStatus exit status of the process @type QProcess.ExitStatus """ if process.timedOut(): msg = self.tr("pipx did not finish within {0} seconds.").format( process.timeoutInterval() // 1_000 ) if callback: callback(False, msg) else: EricMessageBox.critical(None, self.tr("pipx Timeout Error"), msg) elif exitStatus == QProcess.ExitStatus.NormalExit: ioEncoding = Preferences.getSystem("IOEncoding") if process.exitCode() == 0: output = str(process.readAllStandardOutput(), ioEncoding, "replace") if callback: callback(True, output) else: error = str(process.readAllStandardError(), ioEncoding, "replace") msg = self.tr("<p>Message: {0}</p>").format(error) if error else "" if callback: callback( False, self.tr("<p>pipx exited with an error ({0}).</p>{1}").format( process.exitCode(), msg ), ) else: EricMessageBox.critical( None, self.tr("pipx Execution Error"), self.tr("<p>pipx exited with an error ({0}).</p>{1}").format( process.exitCode(), msg ), ) with contextlib.suppress(ValueError): self.__pipxProcesses.remove(process) ############################################################################ ## 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"), parent=self.__ui) 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"), parent=self.__ui) 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"), parent=self.__ui) 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"), parent=self.__ui) 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"), parent=self.__ui) 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"), parent=self.__ui) res = dia.startProcess(self.__getPipxExecutable(), args) if res: dia.exec() 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"), parent=self.__ui) 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"), parent=self.__ui) 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"), parent=self.__ui) 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() def ensurePath(self): """ Public method to ensure that the directory where pipx stores apps is in your PATH environment variable. """ args = ["ensurepath"] dia = PipxExecDialog(self.tr("Ensure PATH Modifications"), parent=self.__ui) res = dia.startProcess(self.__getPipxExecutable(), args) if res: dia.exec() ############################################################################ ## Special methods based on 'runpip' ############################################################################ def checkPackageOutdated(self, package): """ Public method to check, if a given package is outdated. @param package name of the package @type str """ args = ["runpip", package, "list", "--outdated", "--format", "json"] if Preferences.getPip("PipSearchIndex"): indexUrl = Preferences.getPip("PipSearchIndex") + "/simple" args += ["--index-url", indexUrl] self.__runPipxAsyncProcess( args, callback=functools.partial(self.__checkPackageOutdatedCb, package) ) def __checkPackageOutdatedCb(self, package, ok, output): """ Private method handling the pipx process output of a check for an outdated package. @param package name of the package @type str @param ok flag indicating the process ended successfully @type bool @param output output of the pipx process or an error message @type str """ 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), ) self.outdatedPackage.emit(package, "", False) return outdatedList = json.loads(output) # check if the main package is in the list for outdatedPackage in outdatedList: if outdatedPackage["name"] == package: self.outdatedPackage.emit( package, outdatedPackage["latest_version"], len(outdatedList) > 1 ) return self.outdatedPackage.emit(package, "", bool(outdatedList)) def getPackageVersion(self, package): """ Public method to get the version of a package. @param package package name @type str @return package version @rtype str """ packagesList = self.__getPackageDependencies(package=package) for pack in packagesList: if pack["name"] == package: return pack["version"] else: return "" def __getPackageDependencies(self, package, uptodate=False, outdated=False): """ Private method to get a list of dependencies of a given package. @param package name of the package @type str @param uptodate DESCRIPTION (defaults to False) @type TYPE (optional) @param outdated DESCRIPTION (defaults to False) @type TYPE (optional) @return list of dictionaries as returned by 'pip' @rtype list[dict[str: str]] """ if outdated: args = ["runpip", package, "list", "--format", "json", "--outdated"] elif uptodate: args = ["runpip", package, "list", "--format", "json", "--uptodate"] else: args = ["runpip", package, "list", "--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("Get Package Dependencies"), self.tr( "<p>The status of dependencies of package <b>{0}</b> could not" " be determined.</p><p>Reason: {1}</p>" ).format(package, output), ) return [] return json.loads(output) def getOutdatedPackageDependencies(self, package): """ Public method to get the list of outdated package dependencies. @param package name of the package @type str @return list of tuples containing the dependency name, version and latest version @rtype list of tuple of (str, str, str) """ outdatedList = self.__getPackageDependencies(package=package, outdated=True) return [ (d["name"], d["version"], d["latest_version"]) for d in outdatedList if d["name"] != package ] def getUptodatePackageDependencies(self, package): """ Public method to get the list of up-to-date package dependencies. @param package name of the package @type str @return list of tuples containing the dependency name and version @rtype list of tuple of (str, str) """ uptodateList = self.__getPackageDependencies(package=package, uptodate=True) return [(d["name"], d["version"]) for d in uptodateList if d["name"] != package] def getAllPackageDependencies(self, package): """ Public method to get the list of package dependencies. @param package name of the package @type str @return list of tuples containing the dependency name and version @rtype list of tuple of (str, str) """ dependenciesList = self.__getPackageDependencies(package=package) return [ (d["name"], d["version"]) for d in dependenciesList if d["name"] != package ] def upgradePackageDependencies(self, package, dependencies=None): """ Public method to upgrade the dependencies of the given package. @param package name of the package @type str @param dependencies list of dependencies to be upgraded or None to upgrade all outdated dependencies (defaults to None) @type list of str or None (optional) """ if dependencies is None: dependencies = [ d[0] for d in self.getOutdatedPackageDependencies(package=package) ] if dependencies: args = [ "runpip", package, "install", "--upgrade", "--prefer-binary", ] + dependencies dia = PipxExecDialog(self.tr("Upgrade Dependencies"), parent=self.__ui) res = dia.startProcess(self.__getPipxExecutable(), args) if res: dia.exec() else: EricMessageBox.information( self.__ui, self.tr("Upgrade Dependencies"), self.tr("""All dependencies are already up-to-date."""), ) def __getPackageInterpreter(self, package): """ Private method to determine the executable path of the python interpreter of a package. @param package name of the package @type str @return Python interpreter path @rtype str """ from pipx.paths import ctx # noqa: I102 from pipx.venv import Venv # noqa: I102 packagePath = ctx.venvs / package _venv = Venv(packagePath) return str(_venv.python_path) def __getBrokenDependencies(self, dependencies): """ Private method to get a list of broken dependencies. @param dependencies list of dependencies to be checked for being outdated @type list of str @return list of broken dependencies @rtype list of str """ brokenDependecies = [] for dependency in dependencies: if dependency["required_version"].lower() not in ("any", "?"): spec = ( f"=={dependency['required_version']}" if dependency["required_version"][0] in "0123456789" else dependency["required_version"] ) with contextlib.suppress(InvalidSpecifier): specifierSet = SpecifierSet(specifiers=spec) if not specifierSet.contains(dependency["installed_version"]): brokenDependecies.append(f"{dependency['package_name']}{spec}") # recursively add sub-dependencies brokenDependecies.extend( self.__getBrokenDependencies(dependency["dependencies"]) ) return brokenDependecies def repairBrokenDependencies(self, package): """ Public method to get repair broken or unmet package dependencies. @param package name of the package @type str """ dependencies = [] interpreter = self.__getPackageInterpreter(package=package) if interpreter: args = ["-m", "pipdeptree", "--python", interpreter, "--json-tree"] proc = QProcess() proc.start(PythonUtilities.getPythonExecutable(), args) if proc.waitForStarted(15000) and proc.waitForFinished(30000): output = str( proc.readAllStandardOutput(), Preferences.getSystem("IOEncoding"), "replace", ).strip() with contextlib.suppress(json.JSONDecodeError): dependencies = json.loads(output) brokenDependecies = self.__getBrokenDependencies(dependencies) if brokenDependecies: args = [ "runpip", package, "install", "--prefer-binary", ] + brokenDependecies dia = PipxExecDialog( self.tr("Repair Broken Dependencies"), parent=self.__ui ) res = dia.startProcess(self.__getPipxExecutable(), args) if res: dia.exec() else: EricMessageBox.information( self.__ui, self.tr("Repair Broken Dependencies"), self.tr("There are no broken dependencies."), ) else: EricMessageBox.critical( self.__ui, self.tr("Repair Broken Dependencies"), self.tr( "<p>The interpreter for package <b>{0}</b> could not be determined." " Aborting...</p>" ).format(package), )