import json
import boto3
from decimal import Decimal

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('ProcessedItems')

def lambda_handler(event, context):
    """
    Process DynamoDB Stream records from table changes.
    """
    batch_count = 0
    error_count = 0

    for record in event['Records']:
        try:
            event_name = record['eventName']
            dynamodb_data = record['dynamodb']
            seq_num = dynamodb_data['SequenceNumber']

            if event_name == 'INSERT':
                new_image = deserialize(dynamodb_data['NewImage'])
                process_insert(new_image, seq_num)

            elif event_name == 'MODIFY':
                old_image = deserialize(dynamodb_data.get('OldImage', {}))
                new_image = deserialize(dynamodb_data['NewImage'])
                process_modify(old_image, new_image, seq_num)

            elif event_name == 'REMOVE':
                old_image = deserialize(dynamodb_data['OldImage'])
                process_remove(old_image, seq_num)

            batch_count += 1

        except Exception as e:
            print(f"Error processing record: {str(e)}")
            error_count += 1

    print(f"Processed: {batch_count}, Errors: {error_count}")
    return {'processed': batch_count, 'errors': error_count}


def deserialize(image):
    """Convert DynamoDB formatted JSON to regular Python dict."""
    return {k: list(v.values())[0] for k, v in image.items()}


def process_insert(new_image, seq_num):
    """Handle new item creation - index in OpenSearch."""
    item_id = new_image.get('id')
    print(f"INSERT: item={item_id}, seq={seq_num}")
    # Index in OpenSearch for full-text search
    # Notify downstream systems


def process_modify(old_image, new_image, seq_num):
    """Handle item updates - detect specific field changes."""
    item_id = new_image.get('id')
    changed_fields = [
        k for k in new_image
        if k in old_image and old_image[k] != new_image[k]
    ]
    print(f"MODIFY: item={item_id}, changed={changed_fields}, seq={seq_num}")
    # Update search index for changed fields only


def process_remove(old_image, seq_num):
    """Handle item deletion - cleanup downstream resources."""
    item_id = old_image.get('id')
    print(f"REMOVE: item={item_id}, seq={seq_num}")
    # Remove from search index
    # Cleanup related resources

import time
import boto3

dynamodb = boto3.resource('dynamodb')
processed_table = dynamodb.Table('ProcessedEventIds')

def process_record_idempotent(record, handler):
    """Process a stream record with idempotency guarantees."""
    event_id = record['eventID']

    # Check if already processed
    try:
        response = processed_table.get_item(Key={'event_id': event_id})
        if 'Item' in response:
            print(f"Skipping duplicate event: {event_id}")
            return True
    except Exception as e:
        print(f"Error checking idempotency: {str(e)}")

    # Process the record
    handler(record)

    # Mark as processed with conditional write
    try:
        processed_table.put_item(
            Item={
                'event_id': event_id,
                'timestamp': int(time.time()),
                'ttl': int(time.time()) + 86400 * 7  # 7 day TTL
            },
            ConditionExpression='attribute_not_exists(event_id)'
        )
    except dynamodb.meta.client.exceptions.ConditionalCheckFailedException:
        print(f"Concurrent processing detected for: {event_id}")
        return True

    return True

from amazon_kclpy import processor
from amazon_kclpy import kcl
import json
import boto3

class StreamRecordProcessor(processor.RecordProcessorBase):
    """
    Custom DynamoDB Streams consumer using KCL.
    """
    def __init__(self):
        self.batch_size = 100
        self.buffer = []

    def initialize(self, shard_id):
        print(f"Processing shard: {shard_id}")

    def process_records(self, records, checkpointer):
        for record in records:
            event_name = record['dynamodb']['NewImage'] and 'INSERT' or 'REMOVE'
            self.buffer.append({
                'event_id': record['eventID'],
                'event_name': event_name,
                'sequence': record['dynamodb']['SequenceNumber']
            })

        if len(self.buffer) >= self.batch_size:
            self.flush_buffer()
            checkpointer.checkpoint(records[-1]['sequenceNumber'])

    def flush_buffer(self):
        # Process buffered records in bulk
        print(f"Processing batch of {len(self.buffer)} records")
        self.buffer = []

    def shutdown(self, checkpointer, reason):
        if reason == 'TERMINATE':
            self.flush_buffer()
            checkpointer.checkpoint()

import boto3
import json

lambda_client = boto3.client('lambda')

def configure_stream_trigger(function_name, stream_arn):
    """
    Configure Lambda event source mapping for DynamoDB Streams.
    """
    response = lambda_client.create_event_source_mapping(
        FunctionName=function_name,
        EventSourceArn=stream_arn,
        StartingPosition='LATEST',
        BatchSize=100,
        MaximumBatchingWindowInSeconds=60,
        BisectBatchOnFunctionError=True,
        MaximumRetryAttempts=3,
        MaximumRecordAgeInSeconds=86400,  # 24 hours
        TumblingWindowInSeconds=300,  # 5-minute windows
        DestinationConfig={
            'OnFailure': {
                'Destination': 'arn:aws:sqs:us-east-1:123456789012:failed-stream-records'
            }
        },
        FunctionResponseTypes=['ReportBatchItemFailures']
    )

    print(f"Event source mapping created: {response['UUID']}")
    return response['UUID']

import json
import base64
import boto3

def lambda_handler(event, context):
    """
    Transform DynamoDB Stream records for Kinesis Data Firehose delivery.
    """
    output = []

    for record in event['Records']:
        # Decode the DynamoDB stream record
        payload = base64.b64decode(record['data']).decode('utf-8')
        dynamodb_record = json.loads(payload)

        # Transform to target format (Parquet-ready JSON)
        transformed = {
            'event_id': record['recordId'],
            'event_name': dynamodb_record['eventName'],
            'timestamp': dynamodb_record['dynamodb']['ApproximateCreationDateTime'],
            'table_name': dynamodb_record['eventSourceARN'].split('/')[1],
            'keys': dynamodb_record['dynamodb']['Keys'],
            'new_image': dynamodb_record['dynamodb'].get('NewImage', {}),
            'old_image': dynamodb_record['dynamodb'].get('OldImage', {}),
            'sequence_number': dynamodb_record['dynamodb']['SequenceNumber']
        }

        # Add partition key for Firehose S3 prefixing
        transformed['dt'] = __import__('datetime').datetime.fromtimestamp(
            transformed['timestamp']
        ).strftime('%Y-%m-%d')

        output.append({
            'recordId': record['recordId'],
            'result': 'Ok',
            'data': base64.b64encode(
                json.dumps(transformed).encode('utf-8')
            ).decode('utf-8')
        })

    return {'records': output}

import boto3
import json

dynamodb_client = boto3.client('dynamodb')

def setup_global_tables():
    """
    Configure DynamoDB Global Tables for cross-region replication.
    """
    # Global Tables use DynamoDB Streams under the hood
    # This is the managed CDC pattern for multi-region replication
    response = dynamodb_client.create_table(
        TableName='orders-global',
        KeySchema=[
            {'AttributeName': 'order_id', 'KeyType': 'HASH'},
            {'AttributeName': 'timestamp', 'KeyType': 'RANGE'}
        ],
        AttributeDefinitions=[
            {'AttributeName': 'order_id', 'AttributeType': 'S'},
            {'AttributeName': 'timestamp', 'AttributeType': 'N'}
        ],
        BillingMode='PAY_PER_REQUEST',
        GlobalSecondaryIndexes=[
            {
                'IndexName': 'customer-index',
                'KeySchema': [
                    {'AttributeName': 'customer_id', 'KeyType': 'HASH'},
                    {'AttributeName': 'order_date', 'KeyType': 'RANGE'}
                ],
                'Projection': {'ProjectionType': 'ALL'}
            }
        ],
        StreamSpecification={
            'StreamEnabled': True,
            'StreamViewType': 'NEW_AND_OLD_IMAGES'
        }
    )

    # Enable Global Tables replication
    dynamodb_client.update_table(
        TableName='orders-global',
        ReplicaUpdates=[
            {
                'Create': {
                    'RegionName': 'us-west-2',
                    'KMSMasterKeyId': 'alias/aws/dynamodb'
                }
            },
            {
                'Create': {
                    'RegionName': 'eu-west-1',
                    'KMSMasterKeyId': 'alias/aws/dynamodb'
                }
            }
        ]
    )

    print("Global Tables configured for multi-region replication")

-- Query stream records in Athena for analytics
SELECT
    eventname,
    date_format(
        from_unixtime(dynamodb.approximatecreationdatetime),
        '%Y-%m-%d %H:00:00'
    ) as event_hour,
    COUNT(*) as event_count,
    COUNT(DISTINCT dynamodb.keys['id'].s) as unique_items
FROM dynamodb_streams_table
WHERE dt = '2025-01-15'
GROUP BY 1, 2
ORDER BY event_hour, event_count DESC;