Change: CLI framwork for Tyler
This commit is contained in:
parent
aae9aaf03d
commit
c016482852
326
kheops/cli.py
326
kheops/cli.py
@ -5,248 +5,180 @@
|
|||||||
# python3 python_cli.py -vvvv demo
|
# python3 python_cli.py -vvvv demo
|
||||||
# Author: mrjk
|
# Author: mrjk
|
||||||
|
|
||||||
|
import typer
|
||||||
import os
|
import os
|
||||||
import sys
|
import sys
|
||||||
import logging
|
import logging
|
||||||
import argparse
|
|
||||||
import anyconfig
|
import anyconfig
|
||||||
import kheops.app as Kheops
|
import kheops.app as Kheops
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
|
||||||
# Devel tmp
|
# Devel tmp
|
||||||
sys.path.append("/home/jez/prj/bell/training/tiger-ansible/ext/ansible-tree")
|
sys.path.append("/home/jez/prj/bell/training/tiger-ansible/ext/kheops")
|
||||||
|
|
||||||
|
|
||||||
class CmdApp:
|
|
||||||
"""Main CmdApp"""
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
"""Start new App"""
|
|
||||||
|
|
||||||
self.get_args()
|
|
||||||
self.get_logger(verbose=self.args.verbose + 1, logger_name="kheops")
|
|
||||||
self.cli()
|
|
||||||
|
|
||||||
def get_logger(self, logger_name=None, create_file=False, verbose=0):
|
def get_logger(logger_name=None, create_file=False, verbose=0):
|
||||||
"""Create CmdApp logger"""
|
"""Create CmdApp logger"""
|
||||||
|
|
||||||
# Take default app name
|
# Take default app name
|
||||||
if not logger_name:
|
if not logger_name:
|
||||||
logger_name = __name__
|
logger_name = __name__
|
||||||
|
|
||||||
# Manage logging level
|
# Manage logging level
|
||||||
try:
|
try:
|
||||||
loglevel = {
|
loglevel = {
|
||||||
0: logging.ERROR,
|
0: logging.ERROR,
|
||||||
1: logging.WARN,
|
1: logging.WARN,
|
||||||
2: logging.INFO,
|
2: logging.INFO,
|
||||||
3: logging.DEBUG,
|
3: logging.DEBUG,
|
||||||
}[verbose]
|
}[verbose]
|
||||||
except KeyError:
|
except KeyError:
|
||||||
loglevel = logging.DEBUG
|
loglevel = logging.DEBUG
|
||||||
|
|
||||||
# Create logger for prd_ci
|
# Create logger for prd_ci
|
||||||
log = logging.getLogger(logger_name)
|
log = logging.getLogger(logger_name)
|
||||||
log.setLevel(level=loglevel)
|
log.setLevel(level=loglevel)
|
||||||
|
|
||||||
# Formatters
|
# Formatters
|
||||||
format1 = "%(levelname)8s: %(message)s"
|
format1 = "%(levelname)8s: %(message)s"
|
||||||
# format2 = "%(asctime)s.%(msecs)03d|%(name)-16s%(levelname)8s: %(message)s"
|
# format2 = "%(asctime)s.%(msecs)03d|%(name)-16s%(levelname)8s: %(message)s"
|
||||||
# format3 = (
|
# format3 = (
|
||||||
# "%(asctime)s.%(msecs)03d"
|
# "%(asctime)s.%(msecs)03d"
|
||||||
# + " (%(process)d/%(thread)d) "
|
# + " (%(process)d/%(thread)d) "
|
||||||
# + "%(pathname)s:%(lineno)d:%(funcName)s"
|
# + "%(pathname)s:%(lineno)d:%(funcName)s"
|
||||||
# + ": "
|
# + ": "
|
||||||
# + "%(levelname)s: %(message)s"
|
# + "%(levelname)s: %(message)s"
|
||||||
# )
|
# )
|
||||||
tformat1 = "%H:%M:%S"
|
tformat1 = "%H:%M:%S"
|
||||||
# tformat2 = "%Y-%m-%d %H:%M:%S"
|
# tformat2 = "%Y-%m-%d %H:%M:%S"
|
||||||
formatter = logging.Formatter(format1, tformat1)
|
formatter = logging.Formatter(format1, tformat1)
|
||||||
|
|
||||||
# Create console handler for logger.
|
# Create console handler for logger.
|
||||||
stream = logging.StreamHandler()
|
stream = logging.StreamHandler()
|
||||||
stream.setLevel(level=logging.DEBUG)
|
stream.setLevel(level=logging.DEBUG)
|
||||||
stream.setFormatter(formatter)
|
stream.setFormatter(formatter)
|
||||||
log.addHandler(stream)
|
log.addHandler(stream)
|
||||||
|
|
||||||
# Create file handler for logger.
|
# Create file handler for logger.
|
||||||
if isinstance(create_file, str):
|
if isinstance(create_file, str):
|
||||||
handler = logging.FileHandler(create_file)
|
handler = logging.FileHandler(create_file)
|
||||||
handler.setLevel(level=logging.DEBUG)
|
handler.setLevel(level=logging.DEBUG)
|
||||||
handler.setFormatter(formatter)
|
handler.setFormatter(formatter)
|
||||||
log.addHandler(handler)
|
log.addHandler(handler)
|
||||||
|
|
||||||
# Return objects
|
# Return objects
|
||||||
self.log = log
|
return log, loglevel
|
||||||
self.loglevel = loglevel
|
|
||||||
|
|
||||||
def cli(self):
|
|
||||||
"""Main cli command"""
|
|
||||||
|
|
||||||
# Dispatch sub commands
|
|
||||||
if self.args.command:
|
|
||||||
method = "cli_" + str(self.args.command)
|
|
||||||
if hasattr(self, method):
|
|
||||||
getattr(self, method)()
|
|
||||||
else:
|
|
||||||
self.log.error("Subcommand %s does not exists.", self.args.command)
|
|
||||||
else:
|
|
||||||
self.log.error("Missing sub command")
|
|
||||||
self.parser.print_help()
|
|
||||||
|
|
||||||
def get_args(self):
|
|
||||||
"""Prepare command line"""
|
|
||||||
|
|
||||||
# Manage main parser
|
|
||||||
parser = argparse.ArgumentParser(
|
|
||||||
description="Kheops, hierarchical data lookup tool",
|
|
||||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
|
||||||
)
|
|
||||||
parser.add_argument(
|
|
||||||
"-v",
|
|
||||||
"--verbose",
|
|
||||||
action="count",
|
|
||||||
default=int(os.environ.get("KHEOPS_VERBOSE", "0")),
|
|
||||||
help="Increase verbosity (KHEOPS_VERBOSE)",
|
|
||||||
)
|
|
||||||
parser.add_argument(
|
|
||||||
"-c",
|
|
||||||
"--config",
|
|
||||||
default=os.environ.get("KHEOPS_CONFIG", "kheops.yml"),
|
|
||||||
help="Kheops configuration file (KHEOPS_CONFIG)",
|
|
||||||
)
|
|
||||||
# parser.add_argument("help", help="Show usage")
|
|
||||||
subparsers = parser.add_subparsers(
|
|
||||||
title="subcommands", description="valid subcommands", dest="command"
|
|
||||||
)
|
|
||||||
# Manage command: schema
|
|
||||||
add_p = subparsers.add_parser("schema")
|
|
||||||
add_p = subparsers.add_parser("gen_doc")
|
|
||||||
|
|
||||||
# Manage command: lookup2
|
|
||||||
add_p = subparsers.add_parser("lookup")
|
|
||||||
add_p.add_argument(
|
|
||||||
"-n",
|
|
||||||
"--namespace",
|
|
||||||
help="Namespace name (KHEOPS_NAMESPACE)",
|
|
||||||
default=os.environ.get("KHEOPS_NAMESPACE", "default"),
|
|
||||||
)
|
|
||||||
add_p.add_argument(
|
|
||||||
"-f", "--file", help="File with params as dict. Can be stdin - ."
|
|
||||||
)
|
|
||||||
add_p.add_argument(
|
|
||||||
"-e", "--scope", dest="scope_param", action="append", default=[]
|
|
||||||
)
|
|
||||||
add_p.add_argument("-p", "--policy")
|
|
||||||
add_p.add_argument("-t", "--trace", action="store_true")
|
|
||||||
add_p.add_argument("-x", "--explain", action="store_true")
|
|
||||||
add_p.add_argument(
|
|
||||||
"-o",
|
|
||||||
"--format",
|
|
||||||
choices=["yaml", "json", "xml", "ini", "toml"],
|
|
||||||
default="yaml",
|
|
||||||
help="Output format",
|
|
||||||
)
|
|
||||||
add_p.add_argument("keys", default=None, nargs="*")
|
|
||||||
|
|
||||||
# Manage command: demo
|
|
||||||
add_p = subparsers.add_parser("demo")
|
|
||||||
add_p.add_argument("--env", default=os.environ.get("APP_SETTING", "Unset"))
|
|
||||||
add_p.add_argument("--choice", choices=["choice1", "choice2"], type=str)
|
|
||||||
add_p.add_argument("-s", "--store", action="store_true")
|
|
||||||
add_p.add_argument("-a", "--append", dest="appended", action="append")
|
|
||||||
# add_p.add_argument("--short", default=True, required=True)
|
|
||||||
# add_p.add_argument("argument1")
|
|
||||||
# add_p.add_argument("double_args", nargs=2)
|
|
||||||
add_p.add_argument("nargs", nargs="*")
|
|
||||||
|
|
||||||
# Manage command: subcommand2
|
from enum import Enum
|
||||||
upg_p = subparsers.add_parser("subcommand2")
|
from typing import List, Optional
|
||||||
upg_p.add_argument("name")
|
|
||||||
|
|
||||||
# Register objects
|
app = typer.Typer(help="Khéops, hierarchical key/value store")
|
||||||
self.parser = parser
|
|
||||||
self.args = parser.parse_args()
|
|
||||||
|
|
||||||
def cli_demo(self):
|
|
||||||
"""Display how to use logging"""
|
|
||||||
|
|
||||||
self.log.error("Test Critical message")
|
|
||||||
self.log.warning("Test Warning message")
|
|
||||||
self.log.info("Test Info message")
|
|
||||||
self.log.debug("Command line vars: %s", vars(self.args))
|
|
||||||
|
|
||||||
def cli_lookup(self):
|
log, log_level = get_logger(verbose=1, logger_name="kheops")
|
||||||
"""Lookup database"""
|
|
||||||
|
|
||||||
keys = self.args.keys or [None]
|
|
||||||
|
|
||||||
new_params = {}
|
class OutputFormat(str, Enum):
|
||||||
if self.args.file:
|
yaml = "yaml"
|
||||||
new_params = anyconfig.load(self.args.file, ac_parser="yaml")
|
json = "json"
|
||||||
|
toml = "toml"
|
||||||
|
|
||||||
# Parse cli params
|
@app.command()
|
||||||
for i in self.args.scope_param:
|
def lookup(
|
||||||
ret = i.split("=")
|
ctx: typer.Context,
|
||||||
if len(ret) != 2:
|
|
||||||
raise Exception("Malformed params")
|
|
||||||
new_params[ret[0]] = ret[1]
|
|
||||||
|
|
||||||
self.log.info("CLI: %s with env: %s", keys, new_params)
|
namespace: str = typer.Option("default", "-n",
|
||||||
|
help="Name of the namespace",
|
||||||
|
envvar="KHEOPS_NAMESPACE"),
|
||||||
|
keys: List[str] = typer.Argument(None,
|
||||||
|
help="Key(s) to query" ),
|
||||||
|
scope_param: Optional[List[str]] = typer.Option([], "-e",
|
||||||
|
help="Scope variables, var=value",
|
||||||
|
),
|
||||||
|
file: Path = typer.Option(None, "-f",
|
||||||
|
help="Scope file" ),
|
||||||
|
|
||||||
app = Kheops.Kheops(config=self.args.config, namespace=self.args.namespace)
|
|
||||||
ret = app.lookup2(
|
|
||||||
namespace=self.args.namespace,
|
|
||||||
keys=keys,
|
|
||||||
scope=new_params,
|
|
||||||
trace=self.args.trace,
|
|
||||||
explain=self.args.explain,
|
|
||||||
validate_schema=True,
|
|
||||||
)
|
|
||||||
print(anyconfig.dumps(ret, ac_parser=self.args.format))
|
|
||||||
|
|
||||||
def cli_lookup_OLD(self):
|
|
||||||
"""Display how to use logging"""
|
|
||||||
|
|
||||||
# self.log.debug(f"Command line vars: {vars(self.args)}")
|
format: OutputFormat = typer.Option("yaml",
|
||||||
keys = self.args.key or [None]
|
help="Output format",
|
||||||
|
),
|
||||||
|
trace: bool = typer.Option(False),
|
||||||
|
explain: bool = typer.Option(False, "-X",
|
||||||
|
help="Explain the queries"),
|
||||||
|
):
|
||||||
|
"""Lookup database"""
|
||||||
|
|
||||||
# Parse payload from enf file:
|
keys = keys or [None]
|
||||||
new_params = {}
|
config = str(ctx.obj["kheops"]["config"])
|
||||||
if self.args.file:
|
|
||||||
new_params = anyconfig.load(self.args.file, ac_parser="yaml")
|
|
||||||
|
|
||||||
# Parse cli params
|
new_params = {}
|
||||||
for i in self.args.scope_param:
|
if file:
|
||||||
ret = i.split("=")
|
new_params = anyconfig.load(file, ac_parser="yaml")
|
||||||
if len(ret) != 2:
|
|
||||||
raise Exception("Malformed params")
|
|
||||||
new_params[ret[0]] = ret[1]
|
|
||||||
|
|
||||||
self.log.info("CLI: %s with env: %s", keys, new_params)
|
# Parse cli params
|
||||||
|
for i in scope_param:
|
||||||
|
ret = i.split("=")
|
||||||
|
if len(ret) != 2:
|
||||||
|
raise Exception("Malformed params")
|
||||||
|
new_params[ret[0]] = ret[1]
|
||||||
|
|
||||||
app = Kheops.App(config=self.args.config, namespace=self.args.namespace)
|
log.info("CLI: %s with env: %s", keys, new_params)
|
||||||
# for key in keys:
|
|
||||||
ret = app.lookup(
|
|
||||||
keys=keys,
|
|
||||||
scope=new_params,
|
|
||||||
trace=self.args.trace,
|
|
||||||
explain=self.args.explain,
|
|
||||||
validate_schema=True,
|
|
||||||
)
|
|
||||||
print(anyconfig.dumps(ret, ac_parser=self.args.format))
|
|
||||||
|
|
||||||
def cli_schema(self):
|
app = Kheops.Kheops(config=config, namespace=namespace)
|
||||||
"""Display configuration schema"""
|
ret = app.lookup(
|
||||||
|
keys=keys,
|
||||||
|
scope=new_params,
|
||||||
|
trace=trace,
|
||||||
|
explain=explain,
|
||||||
|
validate_schema=True,
|
||||||
|
)
|
||||||
|
typer.echo(anyconfig.dumps(ret, ac_parser=format))
|
||||||
|
|
||||||
app = Kheops.App(config=self.args.config) # , namespace=self.args.namespace)
|
|
||||||
app.dump_schema()
|
|
||||||
|
|
||||||
def cli_gen_doc(self):
|
@app.command()
|
||||||
"""Generate documentation"""
|
def config():
|
||||||
|
typer.echo("Not implemented yet.")
|
||||||
|
|
||||||
|
@app.callback()
|
||||||
|
def main(
|
||||||
|
ctx: typer.Context,
|
||||||
|
verbose: int = typer.Option(0, "--verbose", "-v", count=True),
|
||||||
|
|
||||||
|
config: Path = typer.Option("kheops.yml", "-c",
|
||||||
|
help="Last name of person to greet.",
|
||||||
|
envvar="KHEOPS_CONFIG"),
|
||||||
|
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
Manage users in the awesome CLI app.
|
||||||
|
"""
|
||||||
|
#typer.echo(f"About to execute command: {ctx.invoked_subcommand}")
|
||||||
|
|
||||||
|
log.setLevel(level=verbose)
|
||||||
|
ctx.obj = {
|
||||||
|
"kheops": {
|
||||||
|
"config": config,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
app = Kheops.App(config=self.args.config) # , namespace=self.args.namespace)
|
|
||||||
app.gen_docs()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
CmdApp()
|
|
||||||
|
#typer.run(CmdApp)
|
||||||
|
app()
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user