Script to invoke GitHub Lambdas conditionally

invoke_lambdas_conditionally.py

"""
Script to invoke GitHub Lambdas conditionally
"""
import boto3
import os
import json
import logging
from datetime import datetime
from botocore.exceptions import BotoCoreError, ClientError
from botocore.config import Config

CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
LOG_STORE = os.path.join(CURRENT_DIR, 'log', 'invoke_lambdas.log' )

def init_lambda_client():
    "Initializes and returns a lambda client object"
    try:
        boto3_config = Config(read_timeout = 120)
        client = boto3.client('lambda', config = boto3_config)
        logging.info('Client object created!')        
        return client
    except ClientError as error:
        # Handle client errors (e.g., 400 series HTTP status codes)
        logging.error(f"Client error while initializing lambda client: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error
    except BotoCoreError as error:
        # Handle other BotoCore errors
        logging.error(f"BotoCore error while initializing lambda client: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error
    except Exception as error:
        # Catch any other exceptions
        logging.error(f"An error occurred while initializing lambda client: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error

def parse_lambda_response(lambda_response):
    "Parses lambda response payload from StreamingBody object and returns body of response"
    try:
        # Read the response payload from the StreamingBody
        response_payload = lambda_response.get('Payload').read()
        # Deserialize the response payload(JSON)
        response_data = json.loads(response_payload.decode('utf-8'))
        # Parse the response JSON body
        response_body = json.loads(response_data.get("body"))
        return response_body
    except Exception as error:
        # Catch any other exceptions
        logging.error(f"An error occurred while parsing lambda response: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error

def lambda_workflow():
    "Invokes GitHub lambdas in a sequence according to the run status"
    try:
        client = init_lambda_client()
        # Invoke github stats lambda
        logging.info('Invoke the github stats lambda!')
        response = client.invoke(
            FunctionName = '<stats_lambda>',
            InvocationType = 'RequestResponse'
        )
        response_body = parse_lambda_response(response)
        if (response_body.get("stats_status_flag") == 1):
            logging.info('Invoke the github deltas lambda!')
            response = client.invoke(
                FunctionName = '<deltas_lambda>',
                InvocationType = 'RequestResponse'
                )
            response_body = parse_lambda_response(response)
            if (response_body.get("deltas_status_flag") == 1):
                logging.info('Invoke the github scores lambda!')
                response = client.invoke(
                    FunctionName = '<scores_lambda>',
                    InvocationType = 'RequestResponse'
                    )
                response_body = parse_lambda_response(response)
                if (response_body.get("scores_status_flag") == 1):
                    logging.info('GitHub Lambdas workflow run successful!')
        else:
            logging.error('GitHub Lambdas workflow run failed!')
    except ClientError as error:
        # Handle client errors (e.g., 400 series HTTP status codes)
        logging.error(f"Client error while invoking lambdas: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error
    except BotoCoreError as error:
        # Handle other BotoCore errors
        logging.error(f"BotoCore error while invoking lambdas: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error
    except Exception as error:
        # Catch any other exceptions
        logging.error(f"An error occurred while invoking lambdas: {error}")
        raise Exception(f"Exception encountered and run aborted on {datetime.now()}!") from error

if __name__ == "__main__":
    logging.basicConfig(filename=LOG_STORE, format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO, filemode='a')
    lambda_workflow()