uncloud/uncloud-cli
2
0
Fork 0
Code Issues 3 Pull requests Projects Releases Wiki Activity

refactoring done

Browse source
This commit is contained in:
ahmadbilalkhalid 2019-12-31 19:54:08 +05:00
parent 280043659d
commit dc283251d9
13 changed files with 227 additions and 289 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.gitignore vendored
View file

@ -2,7 +2,7 @@
.vscode .vscode
__pycache__ __pycache__
ucloud_cli.egg-info uncloud_cli.egg-info
build/ build/
dist/ dist/

27
bin/ucloud-cli
View file

@ -1,27 +0,0 @@
#!/usr/bin/env python3
import click
import sys
from ucloud_cli.commands.vm import vm
from ucloud_cli.commands.user import user
from ucloud_cli.commands.host import host
from ucloud_cli.commands.image import image
from ucloud_cli.commands.network import network
from ucloud_cli.helper import exception_handler
@click.group()
def entry_point():
pass
if __name__ == "__main__":
sys.excepthook = exception_handler
entry_point.add_command(vm)
entry_point.add_command(user)
entry_point.add_command(image)
entry_point.add_command(host)
entry_point.add_command(network)
entry_point()

47
bin/uncloud-cli Executable file
View file

@ -0,0 +1,47 @@
#!/usr/bin/env python3
import click
import sys
import logging
from logging.handlers import SysLogHandler
from uncloud_cli.commands.vm import vm
from uncloud_cli.commands.user import user
from uncloud_cli.commands.host import host
from uncloud_cli.commands.image import image
from uncloud_cli.commands.network import network
from uncloud_cli.helper import exception_handler, NoTracebackStreamHandler
@click.group()
def entry_point():
pass
if __name__ == "__main__":
sys.excepthook = exception_handler
# Setting up root logger
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
syslog_handler = SysLogHandler(address='/dev/log')
syslog_handler.setLevel(logging.DEBUG)
syslog_formatter = logging.Formatter('%(pathname)s:%(lineno)d -- %(levelname)-8s %(message)s')
syslog_handler.setFormatter(syslog_formatter)
stream_handler = NoTracebackStreamHandler()
stream_handler.setLevel(logging.INFO)
stream_formatter = logging.Formatter('%(message)s')
stream_handler.setFormatter(stream_formatter)
logger.addHandler(syslog_handler)
logger.addHandler(stream_handler)
entry_point.add_command(vm)
entry_point.add_command(user)
entry_point.add_command(image)
entry_point.add_command(host)
entry_point.add_command(network)
entry_point()

4
conf/ucloud-cli.conf
View file

@ -1,4 +0,0 @@
OTP_NAME=replace_me
OTP_REALM=replace_me
OTP_SEED=replace_me
UCLOUD_API_SERVER=http://[::]:5000

11
setup.py
View file

@ -1,14 +1,12 @@
import os
from setuptools import setup, find_packages from setuptools import setup, find_packages
with open("README.md", "r") as fh: with open("README.md", "r") as fh:
long_description = fh.read() long_description = fh.read()
setup(name='ucloud_cli', setup(name='uncloud_cli',
version='0.1', version='0.1',
description='A utility to interact with ucloud server.', description='A utility to interact with uncloud server.',
url='https://code.ungleich.ch/ucloud/ucloud-cli', url='https://code.ungleich.ch/uncloud/uncloud-cli',
long_description=long_description, long_description=long_description,
long_description_content_type='text/markdown', long_description_content_type='text/markdown',
classifiers=[ classifiers=[
@ -26,6 +24,5 @@ setup(name='ucloud_cli',
'pyotp', 'pyotp',
'click' 'click'
], ],
scripts=['bin/ucloud-cli'], scripts=['bin/uncloud-cli'],
data_files=[(os.path.expanduser('~/ucloud/'), ['conf/ucloud-cli.conf'])],
zip_safe=False) zip_safe=False)

67
uncloud_cli/commands/helper.py
View file

@ -1,16 +1,12 @@
import json import json
import binascii
import click
import requests
from os.path import join as join_path
from pyotp import TOTP from pyotp import TOTP
from uncloud_cli.config import config, config_file
class OTPCredentials:
def __init__(self, name, realm, seed):
self.name = name # type: str
self.realm = realm # type: str
self.seed = seed # type: str
def get_json(self):
return {"name": self.name, "realm": self.realm, "token": TOTP(self.seed).now()}
def load_dump_pretty(content): def load_dump_pretty(content):
@ -18,3 +14,54 @@ def load_dump_pretty(content):
content = content.decode("utf-8") content = content.decode("utf-8")
parsed = json.loads(content) parsed = json.loads(content)
return json.dumps(parsed, indent=4, sort_keys=True) return json.dumps(parsed, indent=4, sort_keys=True)
def make_request(*args, data=None, request_method=requests.post):
r = request_method(
join_path(config['client']['api_server'], *args), json=data
)
print(load_dump_pretty(r.content))
def get_token(ctx, param, value):
if value is not None:
try:
token = TOTP(value).now()
except binascii.Error:
raise click.BadParameter('Please enter the correct seed in {}'.format(config_file))
else:
param.name = 'token'
return token
def add_otp_options(f):
options = [
click.option(
"--name", required=True, default=config['client']['name'],
show_default='name mentioned in {}'.format(config_file)
),
click.option(
"--realm", required=True, default=config['client']['realm'],
show_default='realm mentioned in {}'.format(config_file)
),
click.option(
"--seed", required=True, default=config['client']['seed'],
callback=get_token, show_default='seed mentioned in {}'.format(config_file)
)
]
for opt in reversed(options):
f = opt(f)
return f
def add_vm_options(f):
options = [
click.option('--vm-name', required=True),
click.option('--action', required=True, default=f.__name__)
]
for opt in reversed(options):
f = opt(f)
return f

41
uncloud_cli/commands/host.py
View file

@ -1,9 +1,7 @@
import click import click
import requests import requests
from .helper import OTPCredentials, load_dump_pretty from .helper import add_otp_options, make_request
from uncloud_cli.config import env_vars
from os.path import join as join_path
@click.group() @click.group()
@ -11,28 +9,23 @@ def host():
pass pass
@host.command("create") @host.command('create')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @click.option('--hostname', required=True)
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) @click.option('--cpu', required=True, type=int)
@click.option("--hostname", required=True) @click.option('--ram', required=True)
@click.option("--cpu", required=True, type=int) @click.option('--os-ssd', required=True)
@click.option("--ram", required=True) @click.option('--hdd', default=list(), multiple=True)
@click.option("--os-ssd", required=True) def create(**kwargs):
@click.option("--hdd", default=list(), multiple=True) kwargs['specs'] = {
def create(name, realm, seed, hostname, cpu, ram, os_ssd, hdd): 'cpu': kwargs.pop('cpu'),
data = { 'ram': kwargs.pop('ram'),
**OTPCredentials(name, realm, seed).get_json(), 'os-ssd': kwargs.pop('os_ssd'),
"hostname": hostname, 'hdd': kwargs.pop('hdd')
"specs": {"cpu": cpu, "ram": ram, "os-ssd": os_ssd, "hdd": hdd},
} }
r = requests.post( make_request('host', 'create', data=kwargs)
join_path(env_vars.get("UCLOUD_API_SERVER"), "host", "create"), json=data
)
print(load_dump_pretty(r.content))
@host.command("list") @host.command('list')
def list_host(): def list_host():
r = requests.get(join_path(env_vars.get("UCLOUD_API_SERVER"), "host", "list")) make_request('host', 'list', request_method=requests.get)
print(load_dump_pretty(r.content))

19
uncloud_cli/commands/image.py
View file

@ -1,10 +1,8 @@
from uncloud_cli.commands.helper import load_dump_pretty
from uncloud_cli.config import env_vars
from os.path import join as join_path
import click import click
import requests import requests
from uncloud_cli.commands.helper import make_request
@click.group() @click.group()
def image(): def image():
@ -15,17 +13,12 @@ def image():
@click.option("--public", is_flag=True) @click.option("--public", is_flag=True)
def _list(public): def _list(public):
if public: if public:
r = requests.get(join_path(env_vars.get("UCLOUD_API_SERVER"), "image", "list-public")) make_request('image', 'list-public', request_method=requests.get)
print(load_dump_pretty(r.content))
@image.command("create-from-file") @image.command("create-from-file")
@click.option("--name", required=True) @click.option("--name", required=True)
@click.option("--uuid", required=True) @click.option("--uuid", required=True)
@click.option("--image-store-name", required=True) @click.option("--image-store-name", 'image_store', required=True)
def create_from_file(name, uuid, image_store_name): def create_from_file(**kwargs):
data = {"name": name, "uuid": uuid, "image_store": image_store_name} make_request('image', 'create', data=kwargs)
r = requests.post(
join_path(env_vars.get("UCLOUD_API_SERVER"), "image", "create"), json=data
)
print(load_dump_pretty(r.content))

23
uncloud_cli/commands/network.py
View file

@ -1,6 +1,4 @@
from uncloud_cli.commands.helper import load_dump_pretty, OTPCredentials from uncloud_cli.commands.helper import add_otp_options, make_request
from uncloud_cli.config import env_vars
from os.path import join as join_path
import click import click
import requests import requests
@ -12,20 +10,9 @@ def network():
@network.command("create") @network.command("create")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--network-name", required=True) @click.option("--network-name", required=True)
@click.option("--network-type", required=True) @click.option("--network-type", 'type', required=True)
@click.option("--user", required=True, type=bool, default=False) @click.option("--user", required=True, type=bool, default=False)
def create(name, realm, seed, network_name, network_type, user): def create(**kwargs):
data = { make_request('network', 'create', data=kwargs)
**OTPCredentials(name, realm, seed).get_json(),
"network_name": network_name,
"type": network_type,
"user": user,
}
r = requests.post(
join_path(env_vars.get("UCLOUD_API_SERVER"), "network", "create"), json=data
)
print(load_dump_pretty(r.content))

81
uncloud_cli/commands/user.py
View file

@ -1,9 +1,5 @@
from uncloud_cli.commands.helper import OTPCredentials, load_dump_pretty
from uncloud_cli.config import env_vars
from os.path import join as join_path
import click import click
import requests from uncloud_cli.commands.helper import add_otp_options, make_request
@click.group() @click.group()
@ -12,79 +8,40 @@ def user():
@user.command("files") @user.command("files")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) def list_files(**kwargs):
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) make_request('user', 'files', data=kwargs)
def list_files(name, realm, seed):
data = OTPCredentials(name, realm, seed).get_json()
r = requests.get(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "files"), json=data
)
print(load_dump_pretty(r.content))
@user.command("vms") @user.command("vms")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) def list_vms(**kwargs):
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) make_request('user', 'vms', data=kwargs)
def list_vms(name, realm, seed):
data = OTPCredentials(name, realm, seed).get_json()
r = requests.get(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "vms"), json=data
)
print(load_dump_pretty(r.content))
@user.command("networks") @user.command("networks")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) def list_networks(**kwargs):
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) make_request('user', 'network', data=kwargs)
def list_networks(name, realm, seed):
data = OTPCredentials(name, realm, seed).get_json()
r = requests.get(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "networks"), json=data
)
print(load_dump_pretty(r.content))
@user.command("add-ssh") @user.command("add-ssh")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--key-name", required=True) @click.option("--key-name", required=True)
@click.option("--key", required=True) @click.option("--key", required=True)
def add_ssh(name, realm, seed, key_name, key): def add_ssh(**kwargs):
otp = OTPCredentials(name, realm, seed) make_request('user', 'add-ssh', data=kwargs)
data = {**otp.get_json(), "key_name": key_name, "key": key}
r = requests.post(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "add-ssh"), json=data
)
print(load_dump_pretty(r.content))
@user.command("remove-ssh") @user.command("remove-ssh")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--key-name", required=True) @click.option("--key-name", required=True)
def remove_ssh(name, realm, seed, key_name): def remove_ssh(**kwargs):
otp = OTPCredentials(name, realm, seed) make_request('user', 'remove-ssh', data=kwargs)
data = {**otp.get_json(), "key_name": key_name}
r = requests.get(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "remove-ssh"), json=data
)
print(load_dump_pretty(r.content))
@user.command("get-ssh") @user.command("get-ssh")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--key-name", default="") @click.option("--key-name", default="")
def get_ssh(name, realm, seed, key_name): def get_ssh(**kwargs):
otp = OTPCredentials(name, realm, seed) make_request('user', 'get-ssh', data=kwargs)
data = {**otp.get_json(), "key_name": key_name}
r = requests.get(
join_path(env_vars.get("UCLOUD_API_SERVER"), "user", "get-ssh"), json=data
)
print(load_dump_pretty(r.content))

157
uncloud_cli/commands/vm.py
View file

@ -1,19 +1,7 @@
import click import click
import json
import requests import requests
import subprocess as sp
from uncloud_cli.commands.helper import OTPCredentials, load_dump_pretty from uncloud_cli.commands.helper import add_otp_options, make_request, add_vm_options
from uncloud_cli.config import env_vars
from os.path import join as join_path
def vm_command(command, otp, vm_name, **kwargs):
data = {**otp.get_json(), "vm_name": vm_name, "action": command, **kwargs}
r = requests.post(
join_path(env_vars.get("UCLOUD_API_SERVER"), "vm", "action"), json=data
)
return r
@click.group() @click.group()
@ -21,121 +9,56 @@ def vm():
pass pass
@vm.command("create") @vm.command('create')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @add_vm_options
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) @click.option('--cpu', required=True, type=int)
@click.option("--vm-name", required=True) @click.option('--ram', required=True)
@click.option("--cpu", required=True, type=int) @click.option('--os-ssd', required=True)
@click.option("--ram", required=True) @click.option('--hdd', default=list(), multiple=True)
@click.option("--os-ssd", required=True) @click.option('--image', required=True)
@click.option("--hdd", default=list(), multiple=True) @click.option('--network', default=list(), multiple=True)
@click.option("--image", required=True) def create(**kwargs):
@click.option("--network", default=list(), multiple=True) kwargs['specs'] = {
def create(name, realm, seed, vm_name, cpu, ram, os_ssd, hdd, image, network): 'cpu': kwargs.pop('cpu'),
data = { 'ram': kwargs.pop('ram'),
**OTPCredentials(name, realm, seed).get_json(), 'os-ssd': kwargs.pop('os_ssd'),
"vm_name": vm_name, 'hdd': kwargs.pop('hdd')
"specs": {"cpu": cpu, "ram": ram, "os-ssd": os_ssd, "hdd": hdd},
"network": network,
"image": image,
} }
r = requests.post( make_request('vm', kwargs.pop('action'), data=kwargs)
join_path(env_vars.get("UCLOUD_API_SERVER"), "vm", "create"), json=data
)
print(load_dump_pretty(r.content))
@vm.command("start") @vm.command('start')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @add_vm_options
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) def start(**kwargs):
@click.option("--vm-name", required=True) make_request('vm', 'action', data=kwargs)
@click.option("--in_support_of")
def start(name, realm, seed, vm_name, in_support_of):
r = vm_command(
"start", OTPCredentials(name, realm, seed), vm_name, in_support_of=in_support_of
)
print(load_dump_pretty(r.content))
@vm.command("stop") @vm.command('stop')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @add_vm_options
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) def stop(**kwargs):
@click.option("--vm-name", required=True) make_request('vm', 'action', data=kwargs)
@click.option("--in_support_of")
def stop(name, realm, seed, vm_name, in_support_of):
r = vm_command(
"stop", OTPCredentials(name, realm, seed), vm_name, in_support_of=in_support_of
)
print(load_dump_pretty(r.content))
@vm.command("delete") @vm.command('delete')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @add_vm_options
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) def delete(**kwargs):
@click.option("--vm-name", required=True) make_request('vm', 'action', data=kwargs)
@click.option("--in_support_of")
def delete(name, realm, seed, vm_name, in_support_of):
r = vm_command(
"delete",
OTPCredentials(name, realm, seed),
vm_name,
in_support_of=in_support_of,
)
print(load_dump_pretty(r.content))
@vm.command("status") @vm.command('status')
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM")) @click.option('--vm-name', required=True)
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED")) def status(**kwargs):
@click.option("--vm-name", required=True) make_request('vm', 'status', data=kwargs)
@click.option("--in_support_of")
def status(name, realm, seed, vm_name, in_support_of):
otp = OTPCredentials(name, realm, seed)
data = {**otp.get_json(), "vm_name": vm_name, "in_support_of": in_support_of}
r = requests.get(join_path(env_vars.get("UCLOUD_API_SERVER"), "vm", "status"), json=data)
print(load_dump_pretty(r.content))
@vm.command("migrate") @vm.command("migrate")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME")) @add_otp_options
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--vm-name", required=True) @click.option("--vm-name", required=True)
@click.option("--destination", required=True) @click.option("--destination", required=True)
@click.option("--in_support_of") def vm_migration(**kwargs):
def vm_migration(name, realm, seed, vm_name, destination, in_support_of): make_request('vm', 'migrate', data=kwargs)
otp = OTPCredentials(name, realm, seed)
data = {
**otp.get_json(),
"vm_name": vm_name,
"destination": destination,
"in_support_of": in_support_of,
}
r = requests.post(
join_path(env_vars.get("UCLOUD_API_SERVER"), "vm", "migrate"), json=data
)
print(load_dump_pretty(r.content))
@vm.command("ssh")
@click.option("--name", required=True, default=env_vars.get("OTP_NAME"))
@click.option("--realm", required=True, default=env_vars.get("OTP_REALM"))
@click.option("--seed", required=True, default=env_vars.get("OTP_SEED"))
@click.option("--vm-name", required=True)
@click.option("--in_support_of")
def ssh(name, realm, seed, vm_name, in_support_of):
otp = OTPCredentials(name, realm, seed)
data = {**otp.get_json(), "vm_name": vm_name, "in_support_of": in_support_of}
r = requests.get(join_path(env_vars.get("UCLOUD_API_SERVER"), "vm", "status"), json=data)
try:
_json = json.loads(r.content)
sp.run(['ssh', '-o', 'ConnectTimeout=10',
'root@{}'.format(_json['ip'][0])])
except Exception as err:
print("Some error occurred while accessing VM."
"Make sure VM is running", err)

11
uncloud_cli/config.py
View file

@ -1,9 +1,10 @@
import sys import sys
from os.path import expanduser import configparser
from decouple import Config, RepositoryEnv import os
config_file = os.path.expanduser('~/uncloud/uncloud.conf')
try: try:
env_vars = Config(RepositoryEnv(expanduser("~/uncloud/uncloud-cli.conf"))) config = configparser.ConfigParser()
config.read(config_file)
except Exception as err: except Exception as err:
print(err) sys.exit(err)
sys.exit(1)

24
uncloud_cli/helper.py
View file

@ -1,5 +1,7 @@
import os import os
import sys import sys
import logging
import click
def exception_handler(exception_type, exception, traceback): def exception_handler(exception_type, exception, traceback):
@ -7,3 +9,25 @@ def exception_handler(exception_type, exception, traceback):
sys.__excepthook__(exception_type, exception, traceback) sys.__excepthook__(exception_type, exception, traceback)
else: else:
print("%s: %s" % (exception_type.__name__, exception)) print("%s: %s" % (exception_type.__name__, exception))
class NoTracebackStreamHandler(logging.StreamHandler):
def handle(self, record):
info, cache = record.exc_info, record.exc_text
record.exc_info, record.exc_text = None, None
if record.levelname in ["WARNING", "WARN"]:
click.echo(click.style('', fg='yellow', bold=True, reset=False), nl=False)
elif record.levelname == "ERROR":
click.echo(click.style('', fg='red', bold=True, reset=False), nl=False)
elif record.levelname == "INFO":
click.echo(click.style('', fg='green', bold=True, reset=False), nl=False)
elif record.levelname == "CRITICAL":
click.echo(click.style('', fg='cyan', bold=True, reset=False), nl=False)
try:
super().handle(record)
finally:
record.exc_info = info
record.exc_text = cache
click.echo(click.style('', 'reset'), nl=False)