dynamodb-eventbus-stream-processor.py

from datetime import datetime
import json
import os

import boto3

EVENT_BUS = os.getenv("EVENT_BUS")

events_client = boto3.client("events")


def lambda_handler(event, context):
    """This Lambda function takes DynamoDB stream events and publishes them to an
    EventBridge EventBus in batches (DynamoDB streams can be submitted in batches of a
    maximum of 10).
    """
    events_to_put = []

    for record in event["Records"]:
        print(f"Event: {record['eventName']}/{record['eventID']}")
        table_arn, _ = record["eventSourceARN"].split("/stream")
        events_to_put.append(
            {
                "Time": datetime.utcfromtimestamp(
                    record["dynamodb"]["ApproximateCreationDateTime"]
                ),
                "Source": "my-service.database",
                "Resources": [table_arn],
                "DetailType": record["eventName"],
                "Detail": json.dumps(record),  # Gotcha here: Decimal() objects require handling
                "EventBusName": EVENT_BUS,
            }
        )

    events_client.put_events(Entries=events_to_put)
    return "ok"