Pemicu dan Integrasi Lambda

Events Bring Lambda to Life

How Triggers Work: Push vs. Pull

S3 Event Triggers: React to Storage

S3 Trigger Demo: Logging Uploads

import json

def lambda_handler(event, context):
    """
    Simulated Lambda handler for S3 events.
    Logs the details of the S3 event.
    """
    print("--- S3 Event Received ---")
    for record in event.get('Records', []):
        bucket_name = record['s3']['bucket']['name']
        object_key = record['s3']['object']['key']
        event_name = record['eventName']
        print(f"Bucket: {bucket_name}")
        print(f"Object: {object_key}")
        print(f"Event: {event_name}")
    print("-------------------------")
    return {
        'statusCode': 200,
        'body': json.dumps('S3 event logged!')
    }

if __name__ == "__main__":
    # Simulate an S3 PUT event
    mock_event = {
        "Records": [
            {
                "eventSource": "aws:s3",
                "eventName": "ObjectCreated:Put",
                "s3": {
                    "bucket": {"name": "my-cool-bucket"},
                    "object": {"key": "my-new-file.txt"}
                }
            }
        ]
    }
    lambda_handler(mock_event, None)

DynamoDB Streams: Database Changes

DynamoDB Stream Demo: Processing Updates

import json

def lambda_handler(event, context):
    """
    Simulated Lambda handler for DynamoDB Stream events.
    Logs the details of the changed records.
    """
    print("--- DynamoDB Stream Event Received ---")
    for record in event.get('Records', []):
        event_name = record['eventName']
        keys = record['dynamodb']['Keys']
        new_image = record['dynamodb'].get('NewImage', {})
        print(f"Event: {event_name}")
        print(f"Keys: {json.dumps(keys)}")
        print(f"New Image: {json.dumps(new_image)}")
    print("------------------------------------")
    return {
        'statusCode': 200,
        'body': json.dumps('DynamoDB stream processed!')
    }

if __name__ == "__main__":
    # Simulate a DynamoDB INSERT event
    mock_event = {
        "Records": [
            {
                "eventID": "1",
                "eventName": "INSERT",
                "dynamodb": {
                    "Keys": {"id": {"S": "123"}},
                    "NewImage": {"id": {"S": "123"}, "name": {"S": "Alice"}},
                    "StreamViewType": "NEW_AND_OLD_IMAGES"
                }
            }
        ]
    }
    lambda_handler(mock_event, None)

API Gateway: Serverless API Endpoints

API Gateway Demo: Your Serverless API

import json

def lambda_handler(event, context):
    """
    Simulated Lambda handler for API Gateway events.
    Returns a simple JSON response.
    """
    print("--- API Gateway Event Received ---")
    print(f"HTTP Method: {event.get('httpMethod')}")
    print(f"Path: {event.get('path')}")
    print("----------------------------------")

    response_body = {
        "message": "Hello from your serverless API!",
        "input": event
    }
    return {
        'statusCode': 200,
        'headers': {
            'Content-Type': 'application/json'
        },
        'body': json.dumps(response_body)
    }

if __name__ == "__main__":
    # Simulate an API Gateway GET request event
    mock_event = {
        "resource": "/",
        "path": "/",
        "httpMethod": "GET",
        "headers": {
            "Accept": "text/html"
        },
        "queryStringParameters": None,
        "pathParameters": None,
        "body": None,
        "isBase64Encoded": False
    }
    response = lambda_handler(mock_event, None)
    print("\n--- API Response ---")
    print(json.dumps(response, indent=2))

More Powerful Integrations

Quick Check

Recap: The Heart of Serverless Workflows