uncloud-cli/uncloud_cli/commands/helper.py

import json
import binascii
import click
import requests

from os.path import join as join_path

from pyotp import TOTP
from uncloud_cli.config import config, config_file


def load_dump_pretty(content):
    if isinstance(content, bytes):
        content = content.decode('utf-8')
    parsed = json.loads(content)
    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.get('client', 'api_server'), *args), json=data
    )
    try:
        print(load_dump_pretty(r.content))
    except Exception:
        print('Error occurred while getting output from api server.')


def get_token(ctx, param, value):
    if value is not None:
        try:
            token = TOTP(value).now()
        except binascii.Error:
            raise click.BadParameter('')
        else:
            param.name = 'token'
            return token


def add_otp_options(f):
    options = [
        click.option(
            '--name', required=True, default=config.get('client', 'name', fallback=''),
            show_default='name mentioned in {}'.format(config_file), prompt=True
            ),
        click.option(
            '--realm', required=True, default=config.get('client', 'realm', fallback=''),
            show_default='realm mentioned in {}'.format(config_file), prompt=True
            ),
        click.option(
            '--seed', required=True, default=config.get('client', 'seed', fallback=''),
            callback=get_token, prompt=True,
            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