wrapper helm cli

helm.py

import json
import logging
import subprocess
from os import environ

class Helm:

    def __init__(
        self, helm='helm', kubeconfig=environ.get('KUBECONFIG', None),
        default_namespace='default', default_chart_dir='', raise_ex_on_err=False ):

        self.helm = helm
        self.kubeconfig = kubeconfig
        self.default_namespace = default_namespace
        self.default_chart_dir = default_chart_dir
        self.raise_ex_on_err = raise_ex_on_err

        logging.basicConfig(level =logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s')

    def _run_command(self, command, **kwargs):

        final_command = list(command)

        if 'kubeconfig' in kwargs and kwargs['kubeconfig']:
            final_command.append(f"--kubeconfig {kwargs['kubeconfig']}")
        else:
            final_command.append(f"--kubeconfig {self.kubeconfig}")

        if 'namespace' in kwargs and kwargs['namespace']:
            final_command.append(f"--namespace {kwargs['namespace']}")
        else:
            final_command.append(f"--namespace {self.default_namespace}")

        if 'wait' in kwargs and kwargs['wait']:
            final_command.append('--wait')

        raise_ex_on_err = self.raise_ex_on_err
        if 'raise_ex_on_err' in kwargs:
            raise_ex_on_err = kwargs['raise_ex_on_err']

        json_mode = False
        if 'json' in kwargs and kwargs['json']:
            final_command.append('-o json')
            json_mode = True

        final_command = ' '.join(final_command)
        logging.debug(f'running command: {final_command}')

        result = subprocess.run(final_command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)

        logging.debug(f"{result.stderr} - {result.stdout}")

        if result.stderr.decode('utf-8') != '':
            err = result.stderr.decode('utf-8')
            if raise_ex_on_err:
                raise ValueError(err)
        else:
            err = None

        data = None
        if not err:
            if json_mode:
                data = json.loads(result.stdout)
            else:
                data = result.stdout.decode('utf-8')

        return data, err

    def search(self, keyword, **kwargs):

        command = [self.helm, 'search', 'repo', keyword]
        return self._run_command(command, json=True, **kwargs)

    def pull(self, chart_name, **kwargs):

        chart_dir = self.default_chart_dir
        if 'chart_dir' in kwargs:
            chart_dir = kwargs['chart_dir']

        command = [self.helm, 'pull', chart_name, '--untar',
                   '--untardir', chart_dir]
        return self._run_command(command, **kwargs)

    def list(self, **kwargs):

        command = [self.helm, 'list']
        return self._run_command(command, json=True, **kwargs)

    def install(self, release_name, chart_name, upgrade=False, **kwargs):

        logging.info("Invoke Func Helm Install/Upgrade")

        chart_dir = self.default_chart_dir
        if 'chart_dir' in kwargs:
            chart_dir = kwargs['chart_dir']

        command = [self.helm]

        if upgrade:
            command.append('upgrade')
        else:
            command.append('install')

        command.append(release_name)

        if '/' in chart_name:
            command.append(chart_name)
        else:
            command.append(f'{chart_dir}/{chart_name}')

        return self._run_command(command, json=True, **kwargs)

    def uninstall(self, release_name, **kwargs):

        command = [self.helm, 'uninstall', release_name]
        return self._run_command(command, **kwargs)

    def template(self, release_name, chart_name, **kwargs):
        logging.info("Invoke Func Helm Template")

        chart_dir = self.default_chart_dir
        if 'chart_dir' in kwargs:
            chart_dir = kwargs['chart_dir']

        command = [self.helm, 'template', release_name]

        if '/' in chart_name:
            command.append(chart_name)
        else:
            command.append(f'{chart_dir}/{chart_name}')

        template = self._run_command(command, json=False, **kwargs)
        self.parser(template)

    def package(self, **kwargs):
        logging.info("Invoke Func Helm Package")

        if 'chart_dir' in kwargs:
            chart_dir = kwargs['chart_dir']

        if 'output_dir' in kwargs:
            output_dir = kwargs['output_dir']
        else:
            output_dir = "."

        command = [self.helm, 'package', chart_dir, '--destination', output_dir]

        package = self._run_command(command, json=False, **kwargs)
        print(package)


    def status(self, release_name, **kwargs):
        logging.info("Invoke Func Helm Status")
        command = [self.helm, 'status', release_name]
        # Not Render Data
        #return self._run_command(command, json=True, **kwargs)
        # Render Data
        status = self._run_command(command, json=True, **kwargs) 
        self.parser(status)

    def get_values(self, release_name, **kwargs):

        command = [self.helm, 'get', 'values', release_name, '--all']
        return self._run_command(command, json=True, **kwargs)

    def show_info(self, chart_name, component=all, **kwargs):

        chart_dir = self.default_chart_dir
        if 'chart_dir' in kwargs:
            chart_dir = kwargs['chart_dir']

        command = [self.helm, 'show', component]

        if '/' in chart_name:
            command.append(chart_name)
        else:
            command.append('%s/%s' % (chart_dir, chart_name))

        return self._run_command(command, **kwargs)

    def repo_list(self, **kwargs):

        command = [self.helm, 'repo', 'list']

        data, err = self._run_command(command, json=True, **kwargs)
        if data is None:
            data = []  # workaround for an Helm bug, fixed in the next release
        return data, err

    def repo_add(self, name, url, username=None, password=None, **kwargs):

        command = [self.helm, 'repo', 'add', name, url]
        if username is not None and password is not None:
            command = command + ['--username', username, '--password', password]
        return self._run_command(command, **kwargs)

    def repo_remove(self, name, **kwargs):

        command = [self.helm, 'repo', 'remove', name]
        return self._run_command(command, **kwargs)

    def repo_update(self, **kwargs):

        command = [self.helm, 'repo', 'update']
        return self._run_command(command, **kwargs)

    def parser(self, rdata):
        for data in rdata:
            if type(data) == dict:
                for element in data.keys():
                    if type(data[element]) == str or type(data[element]) == int:
                        print(f"{element.upper()}: {data[element]}")
                    elif type(data[element]) == dict:
                        line = ""
                        for e in data[element]:
                            line = f"{e}: {data[element][e]} -" 
                        print(line)
                    else:
                        print(type(data[element]))
            elif type(data) == str or type(data) == int:
                print(f"{data}")
            elif data == None:
                pass
            else:
                print(f"Type Unknow {data} - {type(data)}")

if __name__ == '__main__':
    helm = Helm()
    # ACTIONS
    #helm.status(release_name="mysql", namespace="apps")
    #helm.template(release_name="mysql", chart_dir="./charts", chart_name="nginx")
    #helm.install(release_name="nginx", namespace="apps", chart_dir="./charts", chart_name="nginx")
    #helm.uninstall(release_name="nginx", namespace="apps")
    #helm.repo_add(name="nginx", url="http://localhost:60677/")
    #helm.pull(chart_name="k8s/nginx-hpc", chart_dir="./charts")
    #helm.package(chart_dir="./charts/nginx")