
import boto3
import traceback
import json
import os
from datetime import datetime, timedelta

code_pipeline = boto3.client('codepipeline')
sns = boto3.client('sns')

def post_notification(action_state):
    topic_arn = os.environ['CODEPIPELINE_FAILURES_TOPIC_ARN']
    message = json.dumps(action_state)

    sns.publish(TopicArn=topic_arn, Message=message)

def lambda_handler(event, context):
    try:
        print('Checking pipeline state...')

        period_start = datetime.now() - timedelta(seconds=60)

        pipeline_name = os.environ['PIPELINE_NAME']

        pipeline_state = code_pipeline.get_pipeline_state(name=pipeline_name)

        for stage_state in pipeline_state['stageStates']:
            for action_state in stage_state['actionStates']:
                exec = action_state['latestExecution']

                if execution['lastStatusChange'] > period_start:
                    if execution['status'] == 'Failed':
                        post_notification(action_state)

        return '...Done'

    except Exception as e:
        print('Function failed due to exception.')
        print(e)
        traceback.print_exc()
        put_job_failure(job, 'Function exception: ' + str(e))


import boto3
import traceback
import json
import os
from datetime import datetime, timedelta

code_pipeline = boto3.client('codepipeline')
sns = boto3.client('sns')

def post_notification(action_state):
    topic_arn = os.environ['CODEPIPELINE_FAILURES_TOPIC_ARN']
    message = json.dumps(action_state)

    sns.publish(TopicArn=topic_arn, Message=message)

def lambda_handler(event, context):
    try:
        print('Checking pipeline state...')

        pipeline_name = os.environ['PIPELINE_NAME']

        pipeline_state = code_pipeline.get_pipeline_state(name=pipeline_name)

        for stage_state in pipeline_state['stageStates']:
            for action_state in stage_state['actionStates']:
                if 'latestExecution' in action_state:
                    execution = action_state['latestExecution']

                    timezone = execution['lastStatusChange'].tzinfo
                    period_start = datetime.now(timezone) - timedelta(seconds=60)

                    if execution['lastStatusChange'] > period_start:
                        if execution['status'] == 'Failed':
                            post_notification(action_state)

        return '...Done'

    except Exception as e:
        print('Function failed due to exception.')
        print(e)
        traceback.print_exc()

Fix CD fail lambda python
Fix CD fail lambda python

  
  import boto3
  import traceback
  import json
  import os
  from datetime import datetime, timedelta
  
  code_pipeline = boto3.client('codepipeline')
  sns = boto3.client('sns')
  
  def post_notification(action_state):
      topic_arn = os.environ['CODEPIPELINE_FAILURES_TOPIC_ARN']
      message = json.dumps(action_state)
  
      sns.publish(TopicArn=topic_arn, Message=message)
  
  def lambda_handler(event, context):
      try:
          print('Checking pipeline state...')
  
-         period_start = datetime.now() - timedelta(seconds=60)
- 
          pipeline_name = os.environ['PIPELINE_NAME']
  
          pipeline_state = code_pipeline.get_pipeline_state(name=pipeline_name)
  
          for stage_state in pipeline_state['stageStates']:
              for action_state in stage_state['actionStates']:
+                 if 'latestExecution' in action_state:
-                 exec = action_state['latestExecution']
+                     execution = action_state['latestExecution']
  
+                     timezone = execution['lastStatusChange'].tzinfo
+                     period_start = datetime.now(timezone) - timedelta(seconds=60)
+ 
-                 if execution['lastStatusChange'] > period_start:
+                     if execution['lastStatusChange'] > period_start:
-                     if execution['status'] == 'Failed':
+                         if execution['status'] == 'Failed':
-                         post_notification(action_state)
+                             post_notification(action_state)
  
          return '...Done'
  
      except Exception as e:
          print('Function failed due to exception.')
          print(e)
          traceback.print_exc()
-         put_job_failure(job, 'Function exception: ' + str(e))
  
Fix CD fail lambda python
## Code Before:

import boto3
import traceback
import json
import os
from datetime import datetime, timedelta

code_pipeline = boto3.client('codepipeline')
sns = boto3.client('sns')

def post_notification(action_state):
    topic_arn = os.environ['CODEPIPELINE_FAILURES_TOPIC_ARN']
    message = json.dumps(action_state)

    sns.publish(TopicArn=topic_arn, Message=message)

def lambda_handler(event, context):
    try:
        print('Checking pipeline state...')

        period_start = datetime.now() - timedelta(seconds=60)

        pipeline_name = os.environ['PIPELINE_NAME']

        pipeline_state = code_pipeline.get_pipeline_state(name=pipeline_name)

        for stage_state in pipeline_state['stageStates']:
            for action_state in stage_state['actionStates']:
                exec = action_state['latestExecution']

                if execution['lastStatusChange'] > period_start:
                    if execution['status'] == 'Failed':
                        post_notification(action_state)

        return '...Done'

    except Exception as e:
        print('Function failed due to exception.')
        print(e)
        traceback.print_exc()
        put_job_failure(job, 'Function exception: ' + str(e))

## Instruction:
Fix CD fail lambda python
## Code After:

import boto3
import traceback
import json
import os
from datetime import datetime, timedelta

code_pipeline = boto3.client('codepipeline')
sns = boto3.client('sns')

def post_notification(action_state):
    topic_arn = os.environ['CODEPIPELINE_FAILURES_TOPIC_ARN']
    message = json.dumps(action_state)

    sns.publish(TopicArn=topic_arn, Message=message)

def lambda_handler(event, context):
    try:
        print('Checking pipeline state...')

        pipeline_name = os.environ['PIPELINE_NAME']

        pipeline_state = code_pipeline.get_pipeline_state(name=pipeline_name)

        for stage_state in pipeline_state['stageStates']:
            for action_state in stage_state['actionStates']:
                if 'latestExecution' in action_state:
                    execution = action_state['latestExecution']

                    timezone = execution['lastStatusChange'].tzinfo
                    period_start = datetime.now(timezone) - timedelta(seconds=60)

                    if execution['lastStatusChange'] > period_start:
                        if execution['status'] == 'Failed':
                            post_notification(action_state)

        return '...Done'

    except Exception as e:
        print('Function failed due to exception.')
        print(e)
        traceback.print_exc()
