""" Command line interfaces for the registrar
"""
import logging.config
import click
import structlog
import structlog.contextvars
from . import registrar
from .daemon import run_daemon
from .config import RegistrarConfig
[docs]def setup_logging(debug=False):
"""Sets up the logging configuration"""
logging.config.dictConfig(
{
"version": 1,
"disable_existing_loggers": False,
"formatters": {
"json": {
"()": structlog.stdlib.ProcessorFormatter,
"processor": structlog.dev.ConsoleRenderer()
if debug
else structlog.processors.JSONRenderer(),
},
},
"handlers": {
"console": {
"class": "logging.StreamHandler",
"level": "DEBUG" if debug else "INFO",
"formatter": "json",
},
},
"root": {
"handlers": ["console"],
"level": "DEBUG" if debug else "INFO",
},
}
)
structlog.configure(
processors=[
structlog.contextvars.merge_contextvars,
structlog.stdlib.filter_by_level,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.add_logger_name,
structlog.stdlib.add_log_level,
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
context_class=structlog.threadlocal.wrap_dict(dict),
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
@click.group()
@click.option("--config-file", type=click.Path(exists=True))
@click.option("--validate", is_flag=True)
@click.option("--host", type=str)
@click.option("--port", type=int)
@click.option("--debug/--no-debug", is_flag=True)
@click.pass_context
def cli(
ctx, config_file=None, validate=False, host=None, port=None, debug=False
):
"""Entry point for the registar"""
ctx.ensure_object(dict)
setup_logging(debug)
# ensure that ctx.obj exists and is a dict (in case `cli()` is called
# by means other than the `if` block below)
if not config_file:
raise ValueError("Missing --config-file parameter")
with open(config_file, encoding="utf-8") as config_file_:
config = RegistrarConfig.from_file(config_file_, validate)
if host:
config.redis_host = host
if port:
config.redis_port = port
ctx.obj["CONFIG"] = config
@cli.command(help="Run the registrar daemon, attaching to a Redis queue")
@click.pass_context
def daemon(ctx):
""" Run the registrar daemon to listen on the given queues
and execute the (de-)registrations commands.
Examples:
\b
registrar \
--config-file config.yaml \
--validate \
--host redis \
--port 6379 \
--debug \
daemon
"""
config: RegistrarConfig = ctx.obj["CONFIG"]
run_daemon(config)
@cli.command(help="Run a single, one-off registration")
@click.argument("route_name", type=str)
@click.argument("item", type=str)
@click.pass_context
def register(ctx, route_name, item):
""" Registers a single item.
Examples:
\b
registrar \
--config-file config.yaml \
--validate \
--host redis \
--port 6379 \
--debug \
register myroute "{...}"
"""
config: RegistrarConfig = ctx.obj["CONFIG"]
# allow to override config values from CLI
registrar.register(config.routes[route_name], config.sources, item)
@cli.command(help="Run a single, one-off de-registration")
@click.argument("route_name", type=str)
@click.argument("item", type=str)
@click.option("--identifier", is_flag=True)
@click.pass_context
def deregister(ctx, route_name, item, identifier):
""" Registers a single item.
Examples:
\b
registrar \
--config-file config.yaml \
--validate \
--host redis \
--port 6379 \
--debug \
deregister --identifier myroute "someid"
"""
config: RegistrarConfig = ctx["CONFIG"]
registrar.deregister(
config.routes[route_name], config.sources, item, use_id=identifier
)
if __name__ == "__main__":
cli(obj={})