def capture_serverless()

in elasticapm/contrib/serverless/aws.py [0:0]
31 lines of code
14 McCabe index (conditional complexity)

def capture_serverless(func: Optional[callable] = None, **kwargs) -> callable:
    """
    Decorator for instrumenting AWS Lambda functions.

    Example usage:

        from elasticapm import capture_serverless

        @capture_serverless
        def handler(event, context):
            return {"statusCode": r.status_code, "body": "Success!"}

    Please note that when using the APM Layer and setting AWS_LAMBDA_EXEC_WRAPPER this is not required and
    the handler would be instrumented automatically.
    """
    if not func:
        # This allows for `@capture_serverless()` in addition to
        # `@capture_serverless` decorator usage
        return functools.partial(capture_serverless, **kwargs)

    if kwargs:
        warnings.warn(
            PendingDeprecationWarning(
                "Passing keyword arguments to capture_serverless will be deprecated in the next major release."
            )
        )

    name = kwargs.pop("name", None)

    kwargs = prep_kwargs(kwargs)

    global INSTRUMENTED
    client = get_client()
    if not client:
        client = Client(**kwargs)
    if not client.config.debug and client.config.instrument and client.config.enabled and not INSTRUMENTED:
        elasticapm.instrument()
        INSTRUMENTED = True

    @functools.wraps(func)
    def decorated(*args, **kwds):
        if len(args) >= 2:
            # Saving these for request context later
            event, context = args[0:2]
        else:
            event, context = {}, {}

        if not client.config.debug and client.config.instrument and client.config.enabled:
            with _lambda_transaction(func, name, client, event, context) as sls:
                sls.response = func(*args, **kwds)
                return sls.response
        else:
            return func(*args, **kwds)

    return decorated