AWS Serverless Function Workflow

import json
import boto3

sns_client = boto3.client('sns', region_name='us-west-2')
SNS_TOPIC_ARN = 'arn:aws:sns:us-west-2:123456789012:my-topic'

def lambda_handler(event, context):
    try:
        for record in event['Records']:
            if record['eventName'] in ['INSERT', 'MODIFY']:
                new_item = record['dynamodb']['NewImage']
                message = {
                    'item_id': new_item['id']['S'],
                    'data': new_item['data']['S']
                }
                sns_client.publish(
                    TopicArn=SNS_TOPIC_ARN,
                    Message=json.dumps(message)
                )
                print(f"Published to SNS: {message}")
        return {
            'statusCode': 200,
            'body': json.dumps('Processed DynamoDB stream events')
        }
    except Exception as e:
        print(f"Error: {e}")
        return {
            'statusCode': 500,
            'body': json.dumps('Failed to process stream')
        }
                

{
  "Comment": "Orchestrates data processing workflow",
  "StartAt": "ExtractData",
  "States": {
    "ExtractData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-west-2:123456789012:function:ExtractData",
      "Next": "TransformData",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.TooManyRequestsException"],
          "IntervalSeconds": 2,
          "MaxAttempts": 3,
          "BackoffRate": 2.0
        }
      ]
    },
    "TransformData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-west-2:123456789012:function:TransformData",
      "Next": "LoadData",
      "Catch": [
        {
          "ErrorEquals": ["TransformError"],
          "Next": "NotifyError"
        }
      ]
    },
    "LoadData": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-west-2:123456789012:function:LoadData",
      "Next": "NotifySuccess",
      "Retry": [
        {
          "ErrorEquals": ["Lambda.ServiceException"],
          "IntervalSeconds": 1,
          "MaxAttempts": 2,
          "BackoffRate": 1.5
        }
      ]
    },
    "NotifySuccess": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-west-2:123456789012:SuccessTopic",
        "Message": "Workflow completed successfully"
      },
      "End": true
    },
    "NotifyError": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:us-west-2:123456789012:ErrorTopic",
        "Message.$": "$.error"
      },
      "End": true
    }
  }
}
                

provider "aws" {
  region = "us-west-2"
}

resource "aws_lambda_function" "api_handler" {
  function_name = "ApiHandler"
  handler       = "index.handler"
  runtime       = "python3.9"
  role          = aws_iam_role.lambda_role.arn
  filename      = "lambda.zip"
  environment {
    variables = {
      QUEUE_URL = aws_sqs_queue.my_queue.id
    }
  }
  tags = {
    Environment = "production"
  }
}

resource "aws_api_gateway_rest_api" "my_api" {
  name        = "MyApi"
  description = "API for Lambda integration"
}

resource "aws_api_gateway_resource" "my_resource" {
  rest_api_id = aws_api_gateway_rest_api.my_api.id
  parent_id   = aws_api_gateway_rest_api.my_api.root_resource_id
  path_part   = "process"
}

resource "aws_api_gateway_method" "my_method" {
  rest_api_id   = aws_api_gateway_rest_api.my_api.id
  resource_id   = aws_api_gateway_resource.my_resource.id
  http_method   = "POST"
  authorization = "NONE"
}

resource "aws_api_gateway_integration" "lambda_integration" {
  rest_api_id             = aws_api_gateway_rest_api.my_api.id
  resource_id             = aws_api_gateway_resource.my_resource.id
  http_method             = aws_api_gateway_method.my_method.http_method
  integration_http_method = "POST"
  type                    = "AWS_PROXY"
  uri                     = aws_lambda_function.api_handler.invoke_arn
}

resource "aws_lambda_permission" "api_gateway_permission" {
  statement_id  = "AllowAPIGatewayInvoke"
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.api_handler.function_name
  principal     = "apigateway.amazonaws.com"
  source_arn    = "${aws_api_gateway_rest_api.my_api.execution_arn}/*/*"
}

resource "aws_sqs_queue" "my_queue" {
  name = "my-queue"
  tags = {
    Environment = "production"
  }
}

resource "aws_iam_role" "lambda_role" {
  name = "api-handler-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "lambda_policy" {
  name = "api-handler-policy"
  role = aws_iam_role.lambda_role.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "logs:CreateLogGroup",
          "logs:CreateLogStream",
          "logs:PutLogEvents",
          "sqs:SendMessage"
        ]
        Resource = "*"
      }
    ]
  })
}