def update_pipeline_rules()

in lambda/build/lambda_pipeline_change.py [0:0]

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

def update_pipeline_rules(event):
    execution_status = event["detail"]["currentPipelineExecutionStatus"]
    pipeline_execution_arn = event["detail"]["pipelineExecutionArn"]
    stage_name = "Build"
    if execution_status == "Executing":
        logger.info(
            f"Disabling code pipeline: {CODE_PIPELINE_NAME}, stage: {stage_name}"
        )
        response = codepipeline.disable_stage_transition(
            pipelineName=CODE_PIPELINE_NAME,
            stageName=stage_name,
            transitionType="Inbound",
            reason=f"Running SageMaker Pipeline Execution: {pipeline_execution_arn}",
        )
        logger.debug(response)
        for rule_name in [DRIFT_RULE_NAME, SCHEDULE_RULE_NAME]:
            update_cloudwatch_rule(rule_name, enable=False)
        return 200, {"action": "Start"}
    else:
        logger.info(
            f"Enabling code pipeline: {CODE_PIPELINE_NAME}, stage: {stage_name}"
        )
        response = codepipeline.enable_stage_transition(
            pipelineName=CODE_PIPELINE_NAME,
            stageName=stage_name,
            transitionType="Inbound",
        )
        logger.debug(response)
        for rule_name in [DRIFT_RULE_NAME, SCHEDULE_RULE_NAME]:
            update_cloudwatch_rule(rule_name, enable=True)
        return 200, {"action": "Stop", "status": execution_status}