decorators.py

"""CLI - Decorators."""

import functools

import click

from ..core.api.init import initialise_api as _initialise_api
from . import config, utils, validators


def report_retry(seconds, context=None):
    if context == "retry-after":
        click.echo()
        click.echo(
            "Request was throttled (429): Retrying after %(seconds)s second(s) ... "
            % {"seconds": click.style(str(seconds), bold=True)}
        )


def common_package_action_options(f):
    """Add common options for package actions."""

    @click.option(
        "-s",
        "--skip-errors",
        default=False,
        is_flag=True,
        help="Skip/ignore errors when copying packages.",
    )
    @click.option(
        "-W",
        "--no-wait-for-sync",
        default=False,
        is_flag=True,
        help="Don't wait for package synchronization to complete before exiting.",
    )
    @click.option(
        "-I",
        "--wait-interval",
        default=5.0,
        type=float,
        show_default=True,
        help=(
            "The minimum time in seconds to wait between checking sync status after "
            "uploading. This is cumulative, so that status checks happen with less "
            "frequency over time, upto a maximum of 5 minutes of waiting."
        ),
    )
    @click.option(
        "--sync-attempts",
        default=3,
        type=int,
        help="Number of times to attempt package synchronization. If the "
        "package fails the first time, the client will attempt to "
        "automatically resynchronize it.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_config_options(f):
    """Add common CLI config options to commands."""

    @click.option(
        "-C",
        "--config-file",
        envvar="CLOUDSMITH_CONFIG_FILE",
        type=click.Path(dir_okay=True, exists=True, writable=False, resolve_path=True),
        help="The path to your config.ini file.",
    )
    @click.option(
        "--credentials-file",
        envvar="CLOUDSMITH_CREDENTIALS_FILE",
        type=click.Path(dir_okay=True, exists=True, writable=False, resolve_path=True),
        help="The path to your credentials.ini file.",
    )
    @click.option(
        "-P",
        "--profile",
        default=None,
        envvar="CLOUDSMITH_PROFILE",
        help="The name of the profile to use for configuration.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        profile = kwargs.pop("profile")
        config_file = kwargs.pop("config_file")
        creds_file = kwargs.pop("credentials_file")
        opts.load_config_file(path=config_file, profile=profile)
        opts.load_creds_file(path=creds_file, profile=profile)
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_output_options(f):
    """Add common CLI output options to commands."""

    @click.option(
        "-d",
        "--debug",
        default=False,
        is_flag=True,
        help="Produce debug output during processing.",
    )
    @click.option(
        "-F",
        "--output-format",
        default="pretty",
        type=click.Choice(["pretty", "json", "pretty_json"]),
        help="Determines how output is formatted. This is only supported by a "
        "subset of the commands at the moment (e.g. list).",
    )
    @click.option(
        "-v",
        "--verbose",
        is_flag=True,
        default=False,
        help="Produce more output during processing.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        opts.debug = kwargs.pop("debug")
        opts.output = kwargs.pop("output_format")
        opts.verbose = kwargs.pop("verbose")
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_cli_list_options(f):
    """Add common list options to commands."""

    @click.option(
        "-p",
        "--page",
        default=1,
        type=int,
        help="The page to view for lists, where 1 is the first page",
        callback=validators.validate_page,
    )
    @click.option(
        "-l",
        "--page-size",
        default=30,
        type=int,
        help="The amount of items to view per page for lists.",
        callback=validators.validate_page_size,
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def common_api_auth_options(f):
    """Add common API authentication options to commands."""

    @click.option(
        "-k",
        "--api-key",
        hide_input=True,
        envvar="CLOUDSMITH_API_KEY",
        help="The API key for authenticating with the API.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        opts = config.get_or_create_options(ctx)
        opts.api_key = kwargs.pop("api_key")
        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper


def initialise_api(f):
    """Initialise the Cloudsmith API for use."""

    @click.option(
        "--api-host", envvar="CLOUDSMITH_API_HOST", help="The API host to connect to."
    )
    @click.option(
        "--api-proxy",
        envvar="CLOUDSMITH_API_PROXY",
        help="The API proxy to connect through.",
    )
    @click.option(
        "-S",
        "--without-api-ssl-verify",
        default=None,
        is_flag=True,
        envvar="CLOUDSMITH_WITHOUT_API_SSL_VERIFY",
        help="Don't verify the SSL connection for the API. This is dangerous and "
        "should only be used in an old/broken environment that doesn't support the "
        "same secure ciphers as Cloudsmith.",
    )
    @click.option(
        "--api-user-agent",
        envvar="CLOUDSMITH_API_USER_AGENT",
        help="The user agent to use for requests.",
    )
    @click.option(
        "--api-headers",
        envvar="CLOUDSMITH_API_HEADERS",
        help="A CSV list of extra headers (key=value) to send to the API.",
    )
    @click.option(
        "-R",
        "--without-rate-limit",
        default=None,
        is_flag=True,
        help="Don't obey the suggested rate limit interval. The CLI will "
        "otherwise automatically sleep between commands to ensure that you do "
        "not hit the server-side rate limit.",
    )
    @click.option(
        "--rate-limit-warning",
        default=30,
        help="When rate limiting, display information that it is happening "
        "if wait interval is higher than this setting. By default no "
        "information will be printed. Set to zero to always see it.",
    )
    @click.option(
        "--error-retry-max",
        default=6,
        help="The maximum amount of times to retry on errors received from "
        "the API, as determined by the --error-retry-codes parameter.",
    )
    @click.option(
        "--error-retry-backoff",
        default=1.0,
        type=float,
        help="The backoff factor determines how long to wait in seconds "
        "between error retries. The backoff factor is multiplied by the "
        "amount of retries so far. So if 1.0, then the wait is 1.0s then "
        "2.0s, then 4.0s, and so forth.",
    )
    @click.option(
        "--error-retry-codes",
        default="429,500,502,503,504",
        help="The status codes that when received from the API will cause "
        "a retry (if --error-retry-max is > 0). By default this will be for "
        "429, 500, 502, 503 and 504 error codes.",
    )
    @click.pass_context
    @functools.wraps(f)
    def wrapper(ctx, *args, **kwargs):
        # pylint: disable=missing-docstring
        def _set_boolean(name, invert=False):
            value = kwargs.pop(name)
            value = value if value is not None else None
            if value is not None and invert:
                value = not value
            return value

        opts = config.get_or_create_options(ctx)
        opts.api_host = kwargs.pop("api_host")
        opts.api_proxy = kwargs.pop("api_proxy")
        opts.api_ssl_verify = _set_boolean("without_api_ssl_verify", invert=True)
        opts.api_user_agent = kwargs.pop("api_user_agent")
        opts.api_headers = kwargs.pop("api_headers")
        opts.rate_limit = _set_boolean("without_rate_limit", invert=True)
        opts.rate_limit_warning = kwargs.pop("rate_limit_warning")
        opts.error_retry_max = kwargs.pop("error_retry_max")
        opts.error_retry_backoff = kwargs.pop("error_retry_backoff")
        opts.error_retry_codes = kwargs.pop("error_retry_codes")
        opts.error_retry_cb = report_retry

        def call_print_rate_limit_info_with_opts(rate_info):
            utils.print_rate_limit_info(opts, rate_info)

        opts.api_config = _initialise_api(
            debug=opts.debug,
            host=opts.api_host,
            key=opts.api_key,
            proxy=opts.api_proxy,
            ssl_verify=opts.api_ssl_verify,
            user_agent=opts.api_user_agent,
            headers=opts.api_headers,
            rate_limit=opts.rate_limit,
            rate_limit_callback=call_print_rate_limit_info_with_opts,
            error_retry_max=opts.error_retry_max,
            error_retry_backoff=opts.error_retry_backoff,
            error_retry_codes=opts.error_retry_codes,
            error_retry_cb=opts.error_retry_cb,
        )

        kwargs["opts"] = opts
        return ctx.invoke(f, *args, **kwargs)

    return wrapper