ucloud-setup/app/helper.py
2019-09-07 13:10:28 +05:00

198 lines
6.2 KiB
Python

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'], path=path)
def clone_etcd_wrapper(path='.'):
sp.check_output(['git', 'clone',
f'https://code.ungleich.ch/ahmedbilal/etcd3_wrapper'], path=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)