python-shctl-iam/iam/cli_click.py

#!/usr/bin/env python3

import io
import logging
import os
import sys
from enum import Enum
from pprint import pprint
from types import SimpleNamespace

from iam.app import App
from iam.framework import get_app_logger

from . import exceptions as error
from .lib.cli_views import VIEW_FORMATS_NAMES, ViewItem, ViewList
from .lib.click_utils import NestedHelpGroup
# from .cli_views import ViewItem, ViewList, VIEW_FORMATS
from .lib.utils import prune, to_yaml

# import rich
# from rich import box
# from rich.console import Console
# from rich.pretty import pprint
# from rich.prompt import Prompt
# from rich.table import Table


# from rich.pretty import pprint


# Rich loader
# ======================

APP_RICH = True
table_box_params = {}

if not APP_RICH:
    import click

    # Create console
    console = SimpleNamespace(print=print)
else:
    # Import rich
    # Transparent rich proxy for click (help menus)
    import rich_click as click
    from rich_click import RichGroup

    from rich import box #, print
    from rich.console import Console
    # Overrides defaults
    from rich.pretty import pprint

    # Load rich wrappers
    class NestedRichHelpGroup(NestedHelpGroup, RichGroup):
        "Add rich class"

    NestedHelpGroup = NestedRichHelpGroup

    # Create default rich table settings
    table_box_params = {
        "box": box.ASCII2,  # box.MINIMAL
        "min_width": 60,
    }

    # Create a console
    console = Console()


logger = logging.getLogger()
cli_logger = logging.getLogger("iam.cli")


# Global vars
# ======================


# Init app
# ======================

logger = logging.getLogger(__name__)

item_view = ViewItem(output=console.print)
list_view = ViewList(output=console.print)


# General App settings
# ===============================


APP_NAME = "iam"
APP_VERSION = "__version__TOFIX"
APP_EXCEPTION = error.IamException


# See: https://docs.python.org/3.8/library/logging.html#logging-levels
DEFAULT_LOG_LEVEL = 30  # warning

# Get default ident
DEBUG = os.environ.get("IAM_DEBUG", "False")
DEFAULT_IDENT = os.environ.get("IAM_IDENT", os.environ.get("SHELL_IDENT", ""))
DEFAULT_FORMAT = os.environ.get("IAM_FORMAT", "table")
DEFAULT_HELP_OPTIONS=["-h", "--help"]
DEFAULT_SHELL = os.environ.get("SHELL", "bash")

# list_view.formats_enum

# Click helpers
# ===============================

_cmd1_options = [click.option("--format")]

_cmd2_options = [click.option("--cmd2-opt")]

CONTEXT_SETTINGS = dict(
    show_default=True,
    help_option_names=DEFAULT_HELP_OPTIONS,
    auto_envvar_prefix=APP_NAME.upper(),
)


def global_options(function):
    # function = click.option(
    #     "debug", "--debug",
    #     is_flag=True, show_default=False, default=False,
    #     help="Debug"
    # )(function)
    # function = click.option(
    #     "force", "--force", "-f"
    #     is_flag=True, show_default=False, default=False,
    #     help="Force"
    # )(function)
    # function = click.option(
    #     "interactive", "--interactive", "-i"
    #     is_flag=True, show_default=False, default=False,
    #     help="Interactive mode"
    # )(function)
    # function = click.option(
    #     "recursive", "--recursive", "-r"
    #     is_flag=True, show_default=False, default=False,
    #     help="Recursive mode"
    # )(function)

    function = click.option(
        "quiet",
        "--quiet",
        "-q",
        is_flag=True,
        show_default=False,
        default=False,
        help="Suppress output except warnings and errors.",
    )(function)

    function = click.option(
        "log_trace",
        "--trace",
        is_flag=True,
        show_default=False,
        default=False,
        help="Show traceback on errors",
    )(function)

    function = click.option(
        "-v",
        "--verbose",
        count=True,
        type=click.IntRange(0, int(DEFAULT_LOG_LEVEL / 10), clamp=True),
        help="Increase verbosity of output. Can be repeated.",
    )(function)

    return function


def format_options(function):
    # function = click.option(
    #     'dry_run', '-n', '--dry-run',
    #     is_flag=True, show_default=False, default=False,
    #     help="Dry run, do not take any actions."
    #     )(function)

    # Duplicates
    function = click.option(
        "-v",
        "--verbose",
        count=True,
        type=click.IntRange(0, int(DEFAULT_LOG_LEVEL / 10), clamp=True),
        help="Increase verbosity of output. Can be repeated.",
    )(function)

    # Special output options
    function = click.option(
        "fmt_name",
        "-O",
        "--output",
        type=click.Choice(VIEW_FORMATS_NAMES, case_sensitive=False),
        help="Output format",
        default=None,
        show_default=DEFAULT_FORMAT,
    )(function)

    function = click.option(
        "fmt_fields", "-H", "--headers", help="Show specific headers"
    )(function)

    function = click.option("fmt_sort", "-S", "--sort", help="Sort columns")(function)

    return function


# Base Application example
# ===============================

# DEFAULT_CONFIG = os.environ.get("IAM_CONFIG", "MISSING")

DEFAULT_CONFIG = click.get_app_dir(APP_NAME)
# SHOW_DEFAULT_HELP=True


def get_var(name, default=None):
    real_name = f"{APP_NAME.upper()}_{name.upper()}"
    # print ("QUERY", real_name)
    return os.environ.get(real_name, default)


# @click.group(cls=NestedHelpGroup, context_settings=CONTEXT_SETTINGS)
@click.group(cls=NestedHelpGroup
)
@global_options
@format_options
@click.version_option()
@click.option(
    "config",
    "--config",
    "-C",
    # envvar='CONFIG', # multiple=True,
    type=click.Path(),
    default=get_var("CONFIG", DEFAULT_CONFIG),
    show_default=get_var("CONFIG", DEFAULT_CONFIG),
    help="Configuration directory or file",
)
@click.pass_context
def cli_app(
    ctx,
    verbose,
    config: str = DEFAULT_IDENT,
    ident: str = DEFAULT_IDENT,
    quiet=False,
    log_trace=False,
    fmt_fields=None,
    fmt_sort=None,
    fmt_name=None,
):
    """Naval Fate.

    This is the docopt example adopted to Click but with some actual
    commands implemented and not just the empty parsing which really
    is not all that interesting.
    """

    # Calculate log level
    log_level = DEFAULT_LOG_LEVEL - verbose * 10
    log_level = log_level if log_level > 0 else 10

    # Define loggers
    loggers = {
        "iam": {"level": log_level},
        "iam.cli": {"level": "DEBUG", "handlers": ["info"]},
    }

    # Instanciate logger
    get_app_logger(loggers=loggers, level=log_level, colors=True)

    cli_config = SimpleNamespace(
        log_level=log_level,
        fmt_name=fmt_name or DEFAULT_FORMAT,
        ident=ident,
    )
    render_config = {
        "fmt_name": cli_config.fmt_name,
        "table_settings": table_box_params,
    }

    render_item = lambda *args, conf={}, **kwargs: item_view.render(
        *args, conf={**render_config, **prune(conf)}, **kwargs
    )
    render_list = lambda *args, conf={}, **kwargs: list_view.render(
        *args, conf={**render_config, **prune(conf)}, **kwargs
    )

    # Instanciate app
    logger.info(f"Current ident: {ident}")
    ctx.obj = SimpleNamespace(
        app=App(config_path=config, ident=ident),
        cli=cli_config,
        render_item=render_item,
        render_list=render_list,
    )


@cli_app.command()
@click.pass_context
def tests(ctx):
    print("Hello world")

    return

    # pprint(command_tree(cli_app))

    print("TREEE @")
    pprint(command_tree3(cli_app, ctx))

    all_help(cli_app, ctx)

    # subcommand_obj = cli_app.get_command(ctx, subcommand)
    # if subcommand_obj is None:
    #     click.echo("I don't know that command.")
    # else:
    #     pprint (subcommand_obj)
    #     pprint (subcommand_obj.__dict__)
    #     click.echo(subcommand_obj.get_help(ctx))


# Cli Utils
# ===============================


# Dedicated help command
# Source: https://stackoverflow.com/a/53137265
@cli_app.command()
@click.argument("subcommand", nargs=-1)
@click.pass_context
def help(ctx, subcommand):
    "Show a command help"

    subcommand = " ".join(subcommand)
    subcommand_obj = cli_app.get_command(ctx, subcommand)
    if subcommand_obj is None:
        click.echo("I don't know that command.")
    else:
        pprint(subcommand_obj)
        pprint(subcommand_obj.__dict__)
        click.echo(subcommand_obj.get_help(ctx))


# Global commands
# ===============================


@cli_app.command()
@click.pass_context
def dump(ctx):
    ctx.obj.app.user_dump()


# Resource kind management
# ===============================


@cli_app.group("kind")
def cli__kind():
    """Manages Resources Kind."""


@cli__kind.command("list")
@click.argument("filter", required=False)
@format_options
@click.pass_context
def kind_list(ctx, filter, fmt_name=None, fmt_sort=None, fmt_fields=None, verbose=None):
    "List resource kinds"

    # Fetch data
    # filter = name
    resources_kind = ctx.obj.app.catalog.resources_kind
    data = []
    for name, item in sorted(resources_kind.items(), key=lambda key: key):
        if filter and not name.startswith(filter):
            continue
        data.append((name, item.desc))

    # Render data
    columns = ["name", "desc"]
    conf = {
        "table_title": "Resources kinds listing",
        "fmt_name": fmt_name,
        "fmt_sort": fmt_sort,
        "fmt_fields": fmt_fields,
    }
    ctx.obj.render_list(data, columns, conf=conf)


@cli__kind.command("show")
@click.argument("name", required=False)
@format_options
@click.pass_context
def kind_show(ctx, name, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "Show resource kind"

    # Fetch data
    columns = ["name", "desc", "input", "needs", "remap"]
    filter = name
    resources_kind = ctx.obj.app.catalog.resources_kind
    for name, item in sorted(resources_kind.items(), key=lambda key: key):
        if filter and not name.startswith(filter):
            continue

        data = [
            item.name,
            item.desc,
            f"{to_yaml(item.input)}",
            f"{to_yaml(item.needs)}",
            f"{to_yaml(item.remap)}",
        ]

        # Render data
        conf = {
            "table_title": f"Resources kind: {data[0]}",
            "fmt_name": fmt_name,
            "fmt_sort": fmt_sort,
            "fmt_fields": fmt_fields,
        }
        ctx.obj.render_item(data, columns, conf=conf)


# @cli__kind.command("move")
# @click.argument("cli__kind")
# @click.argument("x", type=float)
# @click.argument("y", type=float)
# @click.option("--speed", metavar="KN", default=10, help="Speed in knots.")
# def ship_move(cli__kind, x, y, speed):
#     """Moves SHIP to the new location X,Y."""
#     click.echo(f"Moving cli__kind {cli__kind} to {x},{y} with speed {speed}")


# @cli__kind.command("shoot")
# @click.argument("cli__kind")
# @click.argument("x", type=float)
# @click.argument("y", type=float)
# def ship_shoot(cli__kind, x, y):
#     """Makes SHIP fire to X,Y."""
#     click.echo(f"Ship {cli__kind} fires to {x},{y}")


# Resource res management
# ===============================


@cli_app.group("res")
def cli__res():
    """Manages Resources."""


@cli__res.command("list")
@click.argument("filter", required=False)
# @format_options
@click.pass_context
def res_list(ctx, filter, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "List resource ress"

    # Fetch data
    resources = ctx.obj.app.catalog.resources
    data = []
    for name, res in sorted(resources.items(), key=lambda key: key):
        if filter and not name.startswith(filter):
            continue
        data.append(
            (
                res.get_kind(),
                res.get_name(),
                res.desc,
            )
        )

    # Render data
    columns = ["kind", "name", "desc"]
    conf = {
        "table_title": "Resources listing",
        "fmt_name": fmt_name,
        "fmt_sort": fmt_sort,
        "fmt_fields": fmt_fields,
    }
    ctx.obj.render_list(data, columns, conf=conf)


@cli__res.command("show")
@click.argument("name", required=False)
@format_options
@click.pass_context
def res_show(ctx, name, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "Show resource res"

    # Fetch data
    columns = ["name", "uses", "input"]
    columns_all = columns + ["active", "deps", "missing", "loop_max", "loop", "need"]
    filter = name
    resources = ctx.obj.app.catalog.resources.select("startswith", name)
    for name, item in sorted(resources.items(), key=lambda key: key):
        if filter and not name.startswith(filter):
            continue

        if not all and not item.is_active():
            continue

        data = [
            name,
            to_yaml(item.uses).strip(),
            to_yaml(item.input).strip(),
        ]
        if all:
            data.extend(
                [
                    item.is_active(),
                    ",".join([x.name for x in item.resources_deps]),
                    ",".join(item.resources_missing),
                    item.loop_limit,
                    to_yaml(item.loop).strip(),
                    to_yaml(item.needs).strip(),
                ]
            )

        # Render data
        ctx.obj.render_item(
            data,
            columns_all if all else columns,
            conf={
                "table_title": f"Resources kind: {data[0]}",
                "fmt_name": fmt_name,
                "fmt_sort": fmt_sort,
                "fmt_fields": fmt_fields,
            },
        )


# Resource svc management
# ===============================


@cli_app.group("svc")
def cli__svc():
    """Manages Resources Kind."""


@cli__svc.command("list")
@click.argument("filter", required=False)
@format_options
@click.pass_context
def svc_list(ctx, filter, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "List services"

    columns = ["name", "desc"]
    services = ctx.obj.app.catalog.services
    data = []
    for name, res in sorted(services.items(), key=lambda key: key[1].name):
        if filter and not name.startswith(filter):
            continue

        # if not all and not res.is_active():
        #     continue

        data.append(
            (
                # res.get_kind(),
                # res.get_name(),
                res.name,
                res.desc,
            )
        )

    # Render data
    ctx.obj.render_list(
        data,
        columns,
        conf={
            "table_title": f"Services listing",
            "fmt_name": fmt_name,
            "fmt_sort": fmt_sort,
            "fmt_fields": fmt_fields,
        },
    )


@cli__svc.command("show")
@click.argument("name", required=False)
@format_options
@click.pass_context
def svc_show(ctx, name, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "Show service"

    columns = [
        "name",
        "desc",
        "enabled",
        "input",
        "commands",
        "required_svc",
        "resource_lookup",
        "rest_matches" "dump",
    ]
    columns_all = columns + ["active", "deps", "missing", "loop_max", "loop", "need"]

    output = {}
    services = ctx.obj.app.catalog.services.select("startswith", name)
    for name, svc in services.items():
        assert name == svc.name

        data = [
            svc.name,
            svc.desc,
            svc.enabled,
            svc.input,
            to_yaml(svc.list_cmds()).strip(),
            svc.required_services,
            svc.resources_lookup,
            svc.resources_matches,
            svc.dump_item(),
        ]
        data = [str(item) for item in data]

        ctx.obj.render_item(
            data,
            columns_all if all else columns,
            conf={
                "table_title": f"Service show: {data[0]}",
                "fmt_name": fmt_name,
                "fmt_sort": fmt_sort,
                "fmt_fields": fmt_fields,
            },
        )


@cli__svc.command("commands")
@format_options
@click.pass_context
def svc_commands(ctx, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None):
    "Show service"

    columns = ["service", "name", "type", "desc"]

    data = []
    for svc_name, svc in ctx.obj.app.list_services().items():
        cmds = svc.list_cmds()
        for source in ["shell", "cmds"]:
            items = cmds[source]
            # target = ret[source]
            for cmd_name, conf in items.items():
                # target[cmd_name] = conf

                data.append([svc_name, cmd_name, source, conf])

    # Render data
    ctx.obj.render_list(
        data,
        columns,
        conf={
            "table_title": f"Services commands",
            "fmt_name": fmt_name,
            "fmt_sort": fmt_sort,
            "fmt_fields": fmt_fields,
        },
    )


@cli__svc.command("run_v1")
@click.argument("name")
@click.argument("command")
@format_options
@click.pass_context
def svc_run(
    ctx, name, command, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None
):
    "Show service"

    ret = ctx.obj.app.catalog.services.get(name)
    tmp = ret.run_cmd(command)
    pprint(tmp)


# Shell management
# ======================


@cli_app.group("shell")
def cli__shell():
    """Manages Resources Kind."""


@cli__shell.command("idents")
@format_options
@click.pass_context
def shell_idents(
    ctx, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None, 
):
    "Shell identities"
    columns = ["name"]
    data = [[x] for x in ctx.obj.app.get_idents()]

    # Render data
    ctx.obj.render_list(
        data,
        columns,
        conf={
            "table_title": f"Shell identities",
            "fmt_name": fmt_name,
            "fmt_sort": fmt_sort,
            "fmt_fields": fmt_fields,
        },
    )


@cli__shell.command("order")
@click.option(
    "--reverse", is_flag=True, show_default=False, default=False, help="Reverse order"
)
@format_options
@click.pass_context
def shell_order(
    ctx, fmt_name=None, fmt_fields=None, fmt_sort=None, verbose=None, reverse=False
):
    "Shell loading order"

    columns = ["order", "name"]
    data = []
    services, _ = ctx.obj.app.catalog.services.get_loading_order(reverse=reverse)
    for index, name in enumerate(services):
        # table.add_row(str(index), name)
        data.append([str(index), name])

    # Render data
    ctx.obj.render_list(
        data,
        columns,
        conf={
            "table_title": f"Services commands",
            "fmt_name": fmt_name,
            "fmt_sort": fmt_sort,
            "fmt_fields": fmt_fields,
        },
    )




@cli__shell.command("enable")
@click.option(
    "skip_start", "--no-start", "-b", is_flag=True, show_default=False, default=False, help="Disable background process start"
)
@click.pass_context
def shell_enable(ctx, skip_start=True, verbose=None):
    "Enable identity in shell"

    app = ctx.obj.app
    print(app.shell_enable(run_start=not skip_start))



@cli__shell.command("disable")
@click.option(
    "run_stop", "--kill", "-k", is_flag=True, show_default=False, default=False, help="Kill background process on leave"
)
@click.pass_context
def shell_disable(ctx, run_stop=False, verbose=None):
    "Disable identity in shell"

    ret = ctx.obj.app.shell_disable(run_stop=run_stop)
    print(ret)


@cli__shell.command("kill")
@click.option(
    "all_idents", "--all", "-a", is_flag=True, show_default=False, default=False, help="Kill on all users"
)
@click.pass_context
def shell_kill(ctx, all_idents=False, verbose=None):
    "Disable identity in shell"

    ret = []

    app = ctx.obj.app
    if not all_idents:
        ret.append(app.shell_enable(run_start=False))
        ret.append(app.shell_disable(run_stop=True))

    else:
        for ident in app.idents.names():
            cli_logger.warning(f"Kill ident {ident}")
            app.init_ident(ident)
            ret.append(app.shell_enable(run_start=False))
            ret.append(app.shell_disable(run_stop=True))

    print('\n'.join(ret))



@cli__shell.command("switch")
@click.argument("ident", required=False)
# @format_options
@click.option(
    "skip_start", "--no-start", "-b", is_flag=True, show_default=False, default=False, help="Disable background process start"
)
@click.option(
    "run_stop", "--kill", "-k", is_flag=True, show_default=False, default=False, help="Kill background process on leave"
)

@click.pass_context
def shell_switch(ctx, ident="", run_stop=False, skip_start=True, verbose=None):
    "Enable identity in shell"

    app = ctx.obj.app

    src_ident = app.catalog.ident.name
    dst_ident = ident


    if src_ident == dst_ident:
        print (">&2 echo 'No need to change'")
    else:
        ret = []

        # Disable existing
        # logger.info(f"Disabling ident: {app.ident_name}")
        ret.append(app.shell_disable(run_stop=run_stop))

        # Enable new
        # logger.info(f"Enabling ident: {ident}")
        app.init_ident(dst_ident)
        ret.append(app.shell_enable(run_start=not skip_start))

        # Output
        print('\n'.join(ret))





class TmpGroup(click.Group):
    def format_help(self, ctx, formatter):

        val = "List of available commands!"
        formatter.write(val)



@cli_app.command("run",
    # cls=TmpGroup,
    context_settings=dict(
        ignore_unknown_options=True,
        allow_extra_args=True,
        help_option_names=[]
    )
)
@click.argument('cmd_params', nargs=-1, type=click.UNPROCESSED)
@click.pass_context
def cli_app_run(ctx, cmd_params):

    # print("Will run cmd:", cmd_params)

    # Check first help parameter only
    show_help = False
    show_list = False
    for idx, param in enumerate(cmd_params):
        if idx == 0:
            if param in DEFAULT_HELP_OPTIONS:
                show_help = True
            if param in ["list", "ls"]:
                show_list = True

    if show_help:
        ctx = click.get_current_context()
        click.echo(ctx.get_help())
        return


    # Get instanciated app
    app = ctx.obj.app

    if show_list:
        ret = app.catalog.services.list_cmds()
        data = [[x.name, x.desc] for x in ret]


        # Render data
        ctx.obj.render_list(
            data,
            ["name", "desc"],
            conf={
                "table_title": f"Services commands",
                # "fmt_name": fmt_name,
                # "fmt_sort": fmt_sort,
                # "fmt_fields": fmt_fields,
            },
        )

        return
    
    # Run the output
    ret = app.catalog.run_svc_cmd(cmd=cmd_params)
    print (ret)
    # pprint (ret, expand_all=True)



# @cli_app.command("run2",
#     context_settings=dict(
#         ignore_unknown_options=True,
#         allow_extra_args=True,
#     )
# )
# @click.pass_context
# def cli_app_run(ctx):
#     print("Hello world")
#     pprint (ctx.args)

#     return


@cli__shell.command("install")
@click.argument("shell", 
    required=False, 
    default=DEFAULT_SHELL)
@click.pass_context
def shell_install(ctx, shell=None, verbose=None):
    "Install iam in your shell"

    ret = ctx.obj.app.shell_install(shell=shell)

    # print("-- %< --" * 8)
    print(ret)
    # print("-- %< --" * 8)



# Exception handler
# ===============================
def clean_terminate(err):
    "Terminate nicely the program depending the exception"

    # Choose dead end way
    if APP_EXCEPTION is not None and isinstance(err, APP_EXCEPTION):
        err_name = err.__class__.__name__
        # logger.error(err)
        logger.critical("%s: %s" % (err_name, err))
        sys.exit(1)


# Core application executor
# ===============================


def run():
    "Return a MyApp App instance"

    try:
        cli_app(**CONTEXT_SETTINGS)

    # pylint: disable=broad-except
    except Exception as err:
        clean_terminate(err)

        # Developper catchall
        raise Exception(err) from err
        # logger.error(traceback.format_exc())
        # logger.critical("Uncatched error %s; this may be a bug!", err.__class__)
        # logger.critical("Exit 1 with bugs")
        # sys.exit(1)


if __name__ == "__main__":
    run()