EventBridge Pipesを使ってDynamoDBの変更を検知・通知してみた

import json
import boto3
import os
from datetime import datetime

sns = boto3.client('sns')

# 環境変数からSNSトピックARNを取得
SNS_TOPIC_ARN = os.environ['SNS_TOPIC_ARN']
ALERT_THRESHOLD = int(os.environ.get('ALERT_THRESHOLD', '10'))

def lambda_handler(event, context):
    """
    SQSからのメッセージを処理し、在庫アラートをメール送信
    """

    processed_count = 0
    error_count = 0
    skipped_count = 0

    for record in event['Records']:
        try:
            # SQSメッセージの本文を解析
            message_body = json.loads(record['body'])

            # DynamoDBストリームのレコードを解析
            if 'dynamodb' in message_body:
                result = process_dynamodb_record(message_body)
                if result == 'processed':
                    processed_count += 1
                elif result == 'skipped':
                    skipped_count += 1

        except Exception as e:
            print(f"Error processing record: {str(e)}")
            print(f"Record: {json.dumps(record)}")
            error_count += 1
            # エラーの場合、メッセージはSQSに戻される
            raise e

    return {
        'statusCode': 200,
        'body': json.dumps({
            'processed': processed_count,
            'skipped': skipped_count,
            'errors': error_count
        })
    }

def process_dynamodb_record(record):
    """
    DynamoDBレコードを処理してアラートを送信
    """

    event_name = record.get('eventName', '')

    # 新しいイメージと古いイメージを取得
    new_image = record.get('dynamodb', {}).get('NewImage', {})
    old_image = record.get('dynamodb', {}).get('OldImage', {})

    # 商品情報を抽出
    product_id = new_image.get('product_id', {}).get('S', 'Unknown')
    product_name = new_image.get('product_name', {}).get('S', 'Unknown Product')
    new_stock = int(new_image.get('stock_quantity', {}).get('N', '0'))

    # 在庫数が閾値より多い場合はアラートを送信しない
    if new_stock > ALERT_THRESHOLD:
        print(f"Stock level ({new_stock}) is above threshold ({ALERT_THRESHOLD}). Skipping alert for product {product_id}")
        return 'skipped'

    # 古い在庫数を取得（UPDATE時のみ）
    old_stock = None
    if old_image:
        old_stock = int(old_image.get('stock_quantity', {}).get('N', '0'))

    # アラートメッセージの作成
    alert_message = create_alert_message(
        event_name, 
        product_id, 
        product_name, 
        new_stock, 
        old_stock
    )

    # SNSでメール送信
    send_alert(alert_message, product_id, new_stock)

    return 'processed'

def create_alert_message(event_name, product_id, product_name, new_stock, old_stock):
    """
    アラートメッセージを作成
    """

    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

    if event_name == 'INSERT':
        message = f"""
【在庫アラート - 新規商品登録】

日時: {timestamp}
商品ID: {product_id}
商品名: {product_name}
現在の在庫数: {new_stock}個

⚠️ 在庫数が閾値（{ALERT_THRESHOLD}個）以下です。
早急な在庫補充を検討してください。
"""

    elif event_name == 'MODIFY':
        stock_change = new_stock - old_stock if old_stock else 0
        change_text = f"({stock_change:+d}個)" if old_stock else ""

        message = f"""
【在庫アラート - 在庫数更新】

日時: {timestamp}
商品ID: {product_id}
商品名: {product_name}
以前の在庫数: {old_stock if old_stock else 'N/A'}個
現在の在庫数: {new_stock}個 {change_text}

⚠️ 在庫数が閾値（{ALERT_THRESHOLD}個）以下です。
早急な在庫補充を検討してください。
"""

    else:
        message = f"""
【在庫アラート】

日時: {timestamp}
イベント: {event_name}
商品ID: {product_id}
商品名: {product_name}
現在の在庫数: {new_stock}個

⚠️ 在庫数が閾値（{ALERT_THRESHOLD}個）以下です。
"""

    return message

def send_alert(message, product_id, stock_quantity):
    """
    SNS経由でアラートメールを送信
    """

    subject = f"【重要】在庫アラート - {product_id} (残り{stock_quantity}個)"

    try:
        response = sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject=subject,
            Message=message,
            MessageAttributes={
                'product_id': {
                    'DataType': 'String',
                    'StringValue': product_id
                },
                'stock_quantity': {
                    'DataType': 'Number',
                    'StringValue': str(stock_quantity)
                },
                'alert_type': {
                    'DataType': 'String',
                    'StringValue': 'LOW_STOCK'
                }
            }
        )

        print(f"Alert sent successfully. MessageId: {response['MessageId']}")
        return response

    except Exception as e:
        print(f"Failed to send alert: {str(e)}")
        raise e