def process_sns_message()

in lambdas/functions/auto_approval/lambda.py [0:0]

• 35 lines of code
• 4 McCabe index (conditional complexity)

def process_sns_message(sns_message: SNSEvent.sns_message):
    token = sns_message["approval"]["token"]
    pipeline = sns_message["approval"]["pipelineName"]
    stage = sns_message["approval"]["stageName"]
    approval_action = sns_message["approval"]["actionName"]

    logger.info(f"Processing approval for {pipeline}")

    pipeline_name = "".join(
        [k[2] for k in ndiff(f"sagemaker-{project_id}-", pipeline) if k[0] == "+"]
    )
    try:
        flag = ssm_client.get_parameter(
            Name=f"/sagemaker-{project_name}/{pipeline_name}/AutoApprovalFlag"
        )["Parameter"]["Value"]
        flag = bool(flag)
    except:
        logger.exception("Failed to read SSM parameter, default to Manual Approval")
        return

    if not flag:
        logger.info(f"Automatic approval for {pipeline} disabled")
        return

    try:
        cp_client.put_approval_result(
            pipelineName=pipeline,
            stageName=stage,
            actionName=approval_action,
            result={
                "summary": "Automatically approved by Lambda.",
                "status": "Approved",
            },
            token=token,
        )
        logger.info(f"Automatic approval for {pipeline} successful")
    except:
        logger.exception("Failed to Automatically approve the pipeline")

    return