def track_command()

in samcli/lib/telemetry/metric.py [0:0]

• 46 lines of code
• 18 McCabe index (conditional complexity)

def track_command(func):
    """
    Decorator to track execution of a command. This method executes the function, gathers all relevant metrics,
    reports the metrics and returns.

    If you have a Click command, you can track as follows:

    .. code:: python
        @click.command(...)
        @click.options(...)
        @track_command
        def hello_command():
            print('hello')

    """

    @wraps(func)
    def wrapped(*args, **kwargs):
        exception = None
        return_value = None
        exit_reason = "success"
        exit_code = 0

        duration_fn = _timer()

        ctx = None
        try:
            # we have get_current_context in it's own try/except to catch the RuntimeError for this and not func()
            ctx = Context.get_current_context()
        except RuntimeError:
            LOG.debug("Unable to find Click Context for getting session_id.")

        try:
            if ctx and ctx.exception:
                # re-raise here to handle exception captured in context and not run func()
                raise ctx.exception

            # Execute the function and capture return value. This is returned by the wrapper
            # First argument of all commands should be the Context
            return_value = func(*args, **kwargs)
        except (
            UserException,
            click.Abort,
            click.BadOptionUsage,
            click.BadArgumentUsage,
            click.BadParameter,
            click.UsageError,
        ) as ex:
            # Capture exception information and re-raise it later,
            # so metrics can be sent.
            exception = ex
            # NOTE(sriram-mv): Set exit code to 1 if deemed to be user fixable error.
            exit_code = 1
            if hasattr(ex, "wrapped_from") and ex.wrapped_from:
                exit_reason = ex.wrapped_from
            else:
                exit_reason = type(ex).__name__
        except Exception as ex:
            command = ctx.command_path if ctx else ""
            exception = UnhandledException(command, ex)
            # Standard Unix practice to return exit code 255 on fatal/unhandled exit.
            exit_code = 255
            exit_reason = type(ex).__name__

        if ctx:
            time = duration_fn()

            try:
                # metrics also contain a call to Context.get_current_context, catch RuntimeError
                _send_command_run_metrics(ctx, time, exit_reason, exit_code, **kwargs)
            except RuntimeError:
                LOG.debug("Unable to find Click context when sending metrics to telemetry")

        if exception:
            raise exception  # pylint: disable=raising-bad-type

        return return_value

    return wrapped