AWS Cloud Quest: Orchestrate Serverless Workflows Study Note

This is the DIY challenge of the Orchestrate Serverless Workflows in AWS Cloud Quest.

Lambda comprehend_helper

import os, boto3, logging

# AWS Lambda Function Logging in Python - More info: https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Boto3 - Rekognition Client
# More Info: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/comprehend.html
comprehend_client = boto3.client('comprehend')


def lambda_handler(event, context):
    logger.info(event)
    sentiment = comprehend_client.detect_sentiment(Text=event['content'], LanguageCode='en')['Sentiment']
    return {'sentiment': sentiment}

Lambda process_s3_event

import boto3, json, logging

# AWS Lambda Function Logging in Python - More info: https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Boto3 - s3 Client
# More Info: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rekognition.html
s3 = boto3.client('s3')
# More Info: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html?highlight=stepfunctions
step = boto3.client('stepfunctions')


def lambda_handler(event, context):
    logger.info(event)
    for record in event['Records']:
        # Get the bucket name and key for the new file
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        # Get, read, and split the file into lines
        obj = s3.get_object(Bucket=bucket, Key=key)
        response_metadata = s3.head_object(Bucket=bucket, Key=key)
        logger.info('---- Metadata from S3 ----')
        logger.info(response_metadata)

        if response_metadata.get('Metadata') and response_metadata.get('Metadata').get('message'):
            input_step = {
                "s3_info": {
                    'bucket': bucket,
                    'key': key
                },
                "message": {
                    'content': response_metadata['Metadata']['message']
                }

            }
            logger.info('Will start Step function with Input: ' + json.dumps(input_step))
            step.start_execution(
                stateMachineArn='arn:aws:states:us-east-1:<ACCOUNT_NUMBER>:stateMachine:MyStateMachine',
                # name='string',
                input=json.dumps(input_step)
            )
        else:
            logger.info("No metadata found in S3 image")

Lambda rekognition_helper

import boto3, logging

# AWS Lambda Function Logging in Python - More info: https://docs.aws.amazon.com/lambda/latest/dg/python-logging.html
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Boto3 - Rekognition Client
# More Info: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/rekognition.html
rekognition_client = boto3.client('rekognition')


def lambda_handler(event, context):
    logger.info(event)
    response = rekognition_client.detect_moderation_labels(
        Image={"S3Object": {"Bucket": event['bucket'], "Name": event['key']}})
    logger.info(response)
    moderation_labels = response['ModerationLabels'] if 'ModerationLabels' in response else None
    if not moderation_labels:
        return {'safe_content': True}
    else:
        return {'safe_content': False}

DIY Steps:

1. Create an Amazon SQS queue diy
2. Edit step function SendToQueue state

State Machine JSON

{
  "Comment": "A Hello World example of the Amazon States Language using Pass states",
  "StartAt": "Parallel State",
  "States": {
    "Parallel State": {
      "Type": "Parallel",
      "ResultPath": "$.analysis",
      "Branches": [
        {
          "StartAt": "EvaluateImageContent",
          "States": {
            "EvaluateImageContent": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:<ACCOUNT_NUMBER>:function:rekognition_helper",
              "End": true,
              "InputPath": "$.s3_info",
              "ResultPath": "$.result.image_analysis",
              "OutputPath": "$.result"
            }
          }
        },
        {
          "StartAt": "EvaluateMessageContent",
          "States": {
            "EvaluateMessageContent": {
              "Type": "Task",
              "Resource": "arn:aws:lambda:us-east-1:<ACCOUNT_NUMBER>:function:comprehend_helper",
              "End": true,
              "InputPath": "$.message",
              "ResultPath": "$.result.message_analysis",
              "OutputPath": "$.result"
            }
          }
        }
      ],
      "Next": "ModerateOrNotModerate"
    },
    "ModerateOrNotModerate": {
      "Type": "Choice",
      "Choices": [
        {
          "Or": [
            {
              "Variable": "$.analysis[0].image_analysis.safe_content",
              "BooleanEquals": false
            },
            {
              "Variable": "$.analysis[1].message_analysis.sentiment ",
              "StringEquals": "NEGATIVE"
            },
            {
              "Variable": "$.analysis[1].message_analysis.sentiment ",
              "StringEquals": "NEUTRAL"
            },
            {
              "Variable": "$.analysis[1].message_analysis.sentiment ",
              "StringEquals": "MIXED"
            }
          ],
          "Next": "DoSomeModeration"
        }
      ],
      "Default": "ShowComment"
    },
    "DoSomeModeration": {
      "Type": "Pass",
      "Next": "IntoDynamoDB"
    },
    "ShowComment": {
      "Type": "Pass",
      "Next": "SendToQueue"
    },
    "SendToQueue": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sqs:sendMessage",
      "Next": "IntoDynamoDB",
      "OutputPath": "$",
      "ResultPath": "$.queue_response",
      "Parameters": {
        "QueueUrl": "https://sqs.us-east-1.amazonaws.com/<ACCOUNT_NUMBER>/diy",
        "MessageBody.$": "$"
      }
    },
    "IntoDynamoDB": {
      "Type": "Task",
      "Resource": "arn:aws:states:::dynamodb:putItem",
      "Parameters": {
        "TableName": "comments",
        "Item": {
          "id.$": "$$.Execution.Name",
          "comment.$": "$.message.content",
          "bucket.$": "$.s3_info.bucket",
          "key.$": "$.s3_info.key",
          "safe_content": {
            "BOOL.$": "$.analysis[0].image_analysis.safe_content"
          },
          "sentiment.$": "$.analysis[1].message_analysis.sentiment"
        }
      },
      "End": true
    }
  }
}