134 lines
3.7 KiB
Python
134 lines
3.7 KiB
Python
import subprocess
|
|
import inspect
|
|
from dataclasses import dataclass
|
|
from abc import ABC
|
|
from enum import Enum
|
|
|
|
|
|
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 PipenvOperation(object):
|
|
@staticmethod
|
|
def create(path=".", site_packages=False):
|
|
if site_packages:
|
|
command = f"pipenv --site-packages --python 3.7"
|
|
else:
|
|
command = "pipenv --python 3.7"
|
|
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)
|