import subprocess import subprocess as sp import inspect import os from dataclasses import dataclass from abc import ABC from enum import Enum from pipfile import Pipfile class ResultType(Enum): success = 0 failure = 1 def clone(repo): command = f"git clone {repo}" try: subprocess.check_output(command.split()) except subprocess.CalledProcessError as e: return False else: return True def clone_common(path='.'): sp.check_output(['git', 'clone', f'https://code.ungleich.ch/ungleich-public/ucloud_common'], cwd=path) def clone_etcd_wrapper(path='.'): sp.check_output(['git', 'clone', f'https://code.ungleich.ch/ahmedbilal/etcd3_wrapper'], cwd=path) class Result(object): def __init__(self, _result, _type: ResultType, _op=""): self._type = _type self._result = str(_result) self._op = _op if self._type == ResultType.failure: print(self._op, "failed") def add(self, operation, **kwargs): if self._type == ResultType.success: r = operation(**kwargs) self._type = r._type self._result = r._result return self else: print("Dependency not satisfied") exit(-1) def __repr__(self): return f"{self._type}, {self._result}" class Operation(ABC): pass class GitOperation(object): @staticmethod def clone(url, path=".", arguments=""): command = f"git clone {arguments} {url}" try: output = subprocess.check_output(command.split(), cwd=path) except subprocess.CalledProcessError as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result(output, ResultType.success) class VirtualenvOperation(object): @staticmethod def create(path=".", site_packages=False): if site_packages: command = f"virtualenv ." else: command = "virtualenv --system-site-packages ." try: output = subprocess.check_output(command.split(), cwd=path) except subprocess.CalledProcessError as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result(output, ResultType.success) @staticmethod def install(path=".", package_name=None): if package_name: command = f"pip install {package_name}" else: with open(os.path.join(path, "requirements.txt"), "w") as f: subprocess.Popen("pipenv run pip freeze".split(), cwd=path, stdout=f) command = f"pip install -r requirements.txt" try: output = subprocess.check_output(command.split(), cwd=path) except subprocess.CalledProcessError as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result(output, ResultType.success) def globally_installed_py_packages(): output = subprocess.check_output("pip3 list --format freeze".split(), env={}) output = output.decode("utf-8") output = output.strip() global_packages = output.split("\n") return global_packages class PipenvOperation(object): @staticmethod def create(path=".", site_packages=False): if site_packages: global_packages = globally_installed_py_packages() p = Pipfile.load(filename=os.path.join(path, "Pipfile")) content = "" with open(os.path.join(path, "Pipfile"), "r") as f: content = f.read() for pip_package in p.data["default"]: version = p.data["default"][pip_package] if version == "*": for package in global_packages: package = package.lower() if package.startswith(pip_package.lower()): pkg, pkg_ver = package.split("==") substr = f'{pip_package} = "*"' content = content.replace(substr, f'{pkg} = "=={pkg_ver}"') with open(os.path.join(path, "Pipfile"), "w") as f: f.write(content) command = f"pipenv --site-packages --python 3" else: command = "pipenv --python 3" try: output = subprocess.check_output(command.split(), cwd=path) except subprocess.CalledProcessError as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result(output, ResultType.success) @staticmethod def install(path=".", package_name=None): if package_name: command = f"pipenv install {package_name}" else: command = f"pipenv install" try: output = subprocess.check_output(command.split(), cwd=path) except subprocess.CalledProcessError as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result(output, ResultType.success) class FileOperation(object): @staticmethod def write(path, content, mode="w"): try: with open(path, mode) as file: file.write(content) except Exception as e: return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result("", ResultType.success) def get_distro_name(): distro_name = None with open("/etc/os-release") as f: content = f.read() start = content.find("ID=") + 3 end = content.find("\n", start) distro_name = content[start:end] return distro_name class PackageManagementOperation(object): @staticmethod def install(package_name): try: distro = get_distro_name() if distro == "alpine": command = f"apk add {package_name}" else: assert "Unknown Distro" subprocess.check_output(command.split()) except Exception as e: print(e) return Result(e, ResultType.failure, inspect.currentframe().f_code.co_name) else: return Result("", ResultType.success)