You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
85 lines
3.6 KiB
85 lines
3.6 KiB
import subprocess
|
|
import sys
|
|
from pkg_resources import parse_version
|
|
import re
|
|
|
|
class PackageManager:
|
|
__instance = None
|
|
@staticmethod
|
|
def getInstance():
|
|
""" Static access method. """
|
|
if PackageManager.__instance == None:
|
|
PackageManager()
|
|
return PackageManager.__instance
|
|
def __init__(self):
|
|
""" Virtually private constructor. """
|
|
if PackageManager.__instance != None:
|
|
raise Exception("This class is a singleton!")
|
|
else:
|
|
PackageManager.__instance = self
|
|
self.package_list = self.__get_packages()
|
|
|
|
def __get_packages(self):
|
|
packages = {}
|
|
with subprocess.Popen([sys.executable, "-m", "pip", "freeze"],
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE) as p:
|
|
p.wait()
|
|
installed = list(map(lambda x: x.decode("UTF-8").split("=="), filter(lambda x: b'==' in x, p.stdout.read().splitlines())))
|
|
for name, version in installed:
|
|
packages[name] = parse_version(version)
|
|
return packages
|
|
|
|
def ensure_more(self, package_list, executable=sys.executable):
|
|
to_install = list(filter(lambda x: not self.__has_package(x), package_list))
|
|
if len(to_install) > 0:
|
|
command = [executable, "-m", "pip", "install"] + to_install
|
|
with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as p:
|
|
p.wait()
|
|
if p.returncode != 0:
|
|
sys.stdout.buffer.write(p.stderr.read())
|
|
raise(Exception(f"[PythonRunner] Could not install dependencies ({p.returncode})"))
|
|
self.package_list = self.__get_packages()
|
|
|
|
# Assumption: there are more hits in the long run, than misses
|
|
def ensure(self, package_string, executable=sys.executable):
|
|
if not self.__has_package(package_string):
|
|
command = [executable, "-m", "pip", "install", package_string]
|
|
with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as p:
|
|
p.wait()
|
|
if p.returncode != 0:
|
|
sys.stdout.buffer.write(p.stderr.read())
|
|
raise(Exception(f"[PythonRunner] Could not install dependencies ({p.returncode})"))
|
|
self.package_list = self.__get_packages()
|
|
|
|
def __has_package(self, package_string):
|
|
package_data = re.split("==|>|>=|<|<=", package_string)
|
|
# check in cache
|
|
if package_data[0] in self.package_list:
|
|
# check if version is needed
|
|
if len(package_data) == 2:
|
|
required_version = parse_version(package_data[1])
|
|
installed_version = self.package_list[package_data[0]]
|
|
comparator = package_string.replace(package_data[0], "").replace(package_data[1], "")
|
|
if comparator == "==":
|
|
return required_version == installed_version
|
|
elif comparator == ">":
|
|
return installed_version > required_version
|
|
elif comparator == ">=":
|
|
return installed_version >= required_version
|
|
elif comparator == "<":
|
|
return installed_version < required_version
|
|
elif comparator == "<=":
|
|
return installed_version <= required_version
|
|
else:
|
|
raise Exception(f"Illegal comparator found: {comparator}")
|
|
else:
|
|
return True
|
|
return False
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
p = PackageManager().getInstance()
|
|
|
|
print(p.package_list)
|
|
p.ensure_more(["kubernetes", "minio"])
|