# Copyright (C) 2019-2023 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
import logging
import sys
import warnings
import click
import yaml
CONTEXT_SETTINGS = dict(help_option_names=["-h", "--help"])
logger = logging.getLogger(__name__)
[docs]
class AliasedGroup(click.Group):
"""A simple Group that supports:
* command aliases
* notes related to options
It will also ensure the --help option can always be handled by disabling
the callback of the group when the tool is called with --help or -h
"""
# make this class the default Group, see
# https://click.palletsprojects.com/en/stable/api/#click.Group.group_class
group_class = type
def __init__(self, name=None, commands=None, **attrs):
self.option_notes = attrs.pop("option_notes", None)
self.aliases = {}
super().__init__(name, commands, **attrs)
[docs]
def get_command(self, ctx, cmd_name):
return super().get_command(ctx, self.aliases.get(cmd_name, cmd_name))
[docs]
def add_alias(self, name, alias):
if not isinstance(name, str):
name = name.name
self.aliases[alias] = name
[docs]
def invoke(self, ctx):
# do NOT execute groups callback if this command is called for help;
# group callbacks may try to do stuff (esp. loading and partially checking
# configuration) that we do not want to fail when asking for help...
# We need to restore the actual callback after execution for tests of
# other sw packages not to break; they may test the behavior of their
# cli's '--help' output, which would then prevent any further cli call
# to fail...
orig_cb = ctx.command.callback
if set(ctx.help_option_names) & set(ctx.args):
ctx.command.callback = None
try:
return super().invoke(ctx)
finally:
ctx.command.callback = orig_cb
[docs]
def clean_exit_on_signal(signal, frame):
"""Raise a SystemExit exception to let command-line clients wind themselves
down on exit"""
raise SystemExit(0)
[docs]
def validate_loglevel_params(ctx, param, values):
"""Validate the --log-level parameters, with multiple values."""
if values is None:
return None
from swh.core.logging import BadLogLevel, validate_loglevel
try:
module_log_levels = [validate_loglevel(value) for value in values]
except BadLogLevel as e:
raise click.BadParameter(e)
return module_log_levels
[docs]
def validate_options(ctx, param, values):
if values is None:
return values
options = []
for option in values:
kv = option.split("=", 1)
if len(kv) != 2:
raise ValueError("Invalid option")
k, v = kv
options.append((k.strip(), yaml.safe_load(v)))
return options
[docs]
def setup_config(ctx, config_file, options=None):
from os import environ
from swh.core.config import read as config_read
ctx.ensure_object(dict)
if config_file and ctx.info_name != "swh":
logger.warn(
f"Using --config-file on the {ctx.info_name} command group is deprecated"
)
if "config" not in ctx.obj:
if config_file is None:
config_file = environ.get("SWH_CONFIG_FILENAME")
ctx.obj["config"] = config_read(config_file)
if options is not None:
for k, v in options:
kpath = k.split(".")
cfg = ctx.obj["config"]
for elt in kpath[:-1]:
cfg = cfg.setdefault(elt, {})
cfg[kpath[-1]] = v
[docs]
def show_versions(
ctx: click.core.Context, param: click.core.Parameter, value: bool
) -> None:
if not value or ctx.resilient_parsing:
return
from importlib import metadata as MD
version = MD.version("swh.core")
click.echo(f"swh.core, version {version}")
click.echo("Installed swh components:")
for pdist in sorted(
(x for x in MD.distributions() if x.name.startswith("swh.")),
key=lambda x: x.name,
):
try:
click.echo(f" {pdist.name}, version {pdist.version}")
except MD.PackageNotFoundError:
pass
ctx.exit()
@click.group(
context_settings=CONTEXT_SETTINGS,
cls=AliasedGroup,
option_notes="""\
If both options are present, --log-level values will override the configuration
in --log-config.
The --log-config YAML must conform to the logging.config.dictConfig schema
documented at https://docs.python.org/3/library/logging.config.html.
""",
)
@click.option(
"--log-level",
"-l",
"log_levels",
envvar="SWH_LOG_LEVEL",
default=None,
callback=validate_loglevel_params,
multiple=True,
help=(
"Log level (defaults to INFO). "
"Can override the log level for a specific module, by using the "
"``specific.module:LOGLEVEL`` syntax (e.g. ``--log-level swh.core:DEBUG`` "
"will enable DEBUG logging for swh.core)."
),
)
@click.option(
"--log-config",
default=None,
type=click.Path(exists=True, readable=True),
help="Python yaml logging configuration file.",
)
@click.option(
"--sentry-dsn", default=None, help="DSN of the Sentry instance to report to"
)
@click.option(
"--sentry-debug/--no-sentry-debug",
default=False,
hidden=True,
help="Enable debugging of sentry",
)
@click.option(
"--config-file",
"-C",
default=None,
type=click.Path(exists=True, dir_okay=False),
help="Configuration file.",
)
@click.option(
"--option",
"-o",
"options",
multiple=True,
callback=validate_options,
help=(
"Configuration option; set or override a configuration "
"entry from the configuration file"
),
)
@click.option(
"--versions",
is_flag=True,
expose_value=False,
is_eager=True,
callback=show_versions,
help="Show versions of all loaded swh packages",
)
@click.pass_context
def swh(ctx, log_levels, log_config, sentry_dsn, sentry_debug, config_file, options):
"""Command line interface for Software Heritage."""
import signal
from swh.core.logging import logging_configure
signal.signal(signal.SIGTERM, clean_exit_on_signal)
signal.signal(signal.SIGINT, clean_exit_on_signal)
if sentry_dsn is not None:
from ..sentry import init_sentry
init_sentry(sentry_dsn=sentry_dsn, debug=sentry_debug)
set_default_loglevel = logging_configure(log_levels, log_config)
ctx.ensure_object(dict)
ctx.obj["log_level"] = set_default_loglevel
ctx.__class__.formatter_class = SWHHelpFormatter
setup_config(ctx, config_file, options)
# XXX find a decent way to dump the actual configuration in the logs
# hiding credentials (including in connection string...)
[docs]
def main():
# Even though swh() sets up logging, we need an earlier basic logging setup
# for the next few logging statements
logging.basicConfig()
if sys.version_info < (3, 10):
from backports.entry_points_selectable import entry_points
else:
from importlib.metadata import entry_points
# load plugins that define cli sub commands
for entry_point in entry_points(group="swh.cli.subcommands"):
try:
cmd = entry_point.load()
if isinstance(cmd, click.BaseCommand):
# for BW compat, auto add click commands
warnings.warn(
f"{entry_point.name}: automagic addition of click commands "
f"to the main swh group is deprecated",
DeprecationWarning,
)
swh.add_command(cmd, name=entry_point.name)
# otherwise it's expected to be a module which has been loaded
# it's the responsibility of the click commands/groups in this
# module to transitively have the main swh group as parent.
except Exception as e:
logger.warning("Could not load subcommand %s: %r", entry_point.name, e)
return swh(auto_envvar_prefix="SWH")
if __name__ == "__main__":
main()
