ucloud-setup/app/helper.py

import subprocess
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():
    return clone("https://code.ungleich.ch/ungleich-public/ucloud_common")


def clone_etcd_wrapper():
    return clone("https://code.ungleich.ch/ahmedbilal/etcd3_wrapper")


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="."):
        command = f"git clone {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("pip 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)