def step_function_step()

in source/forecast-shared/shared/helpers.py [0:0]

• 30 lines of code
• 9 McCabe index (conditional complexity)

def step_function_step(f):  # NOSONAR - higher cognitive complexity allowed
    """
    Used to wrap AWS Lambda Functions that produce an AWS Forecast resource status.
    :param f: the function to wrap
    :return: the wrapped function
    """

    @wraps(f)
    def wrapper(event, context):
        try:
            (status, output) = f(event, context)
        except get_forecast_client().exceptions.ResourceInUseException:
            logger.info("resource is currently updating")
            raise ResourcePending
        except get_forecast_client().exceptions.LimitExceededException as limit_exception:  # deal with (retryable) forecast rate limiting
            if "concurrently" in str(limit_exception):
                raise ResourcePending
            elif "dataset import jobs" in str(
                limit_exception
            ):  # "Quota limit of \d+ dataset import jobs has been reached"
                raise ResourcePending
            else:
                raise limit_exception  # reraise - user will have to make changes to forecast or file a quota change request ticket
        else:
            if status.failed:
                raise ResourceFailed
            elif status.updating:
                raise ResourcePending
            elif status.finalized:
                return output
            else:
                logger.critical(
                    "invalid resource detected, status is %s" % status, stack_info=True
                )
                raise ResourceInvalid(f"This should not happen: Status is {status}")

    return wrapper