def lambda_handler()

in src/main.py [0:0]

• 42 lines of code
• 6 McCabe index (conditional complexity)

def lambda_handler(event, context):
    print(json.dumps(event))
    required_param_list = ['cfn_template']

    job_id = event['CodePipeline.job']['id']
    job_data = event['CodePipeline.job']['data']
    artifacts = job_data['inputArtifacts']
    dest_account = event['CodePipeline.job']['accountId']

    try:
        # Set assumed role
        if dest_account != os.environ['CURRENT_ACCOUNT']:
            credentials = assume_role(account_number=dest_account, role_name=os.environ['ROLE'])
            session = boto3_session(credentials=credentials)
        else:
            session = boto3_session()

    except Exception as e:
        logger.error(f"Unable to assume role {os.environ['ROLE']}")
        session = boto3_session(credentials=job_data['artifactCredentials'])
        put_job_failure(job_id=job_id, message=str(e), session=session)

    try:
        # Extract the params
        params = get_user_params(
            job_data=job_data,
            required_param_list=required_param_list
        )

        # Get CloudFormation files from s3
        logger.info("Get CloudFormation Template file from CodePipeline - S3")
        tmp_dirname = tempfile.mkdtemp()
        template_location = download_file_from_pipeline_s3(
            job_data=job_data,
            artifact=artifacts[0],
            file_in_zip=params['cfn_template'],
            download_dir=tmp_dirname
        )

        if template_location:
            scan_results = check_codepipeline_cfn_template(template=template_location)

            if "Failed:" in ",".join(scan_results):
                raise Exception(f"Pipeline Scan {scan_results}")
        else:
            logger.warning('No CloudFormation Template or Parameter file found.')

        put_job_success(job_id=job_id, message='Deployment was successful', session=session)

    except Exception as e:
        logger.info(e)
        put_job_failure(job_id=job_id, message=f"{str(e)}", session=session)

    finally:
        # Clean up the /tmp folder to avoid overlap on subsequent runs
        cleanup_temp(temp_location=tmp_dirname)