Lambda Event Sources — Triggering Functions from AWS Services

DodaTech Updated 2026-06-28 7 min read

In this tutorial, you will learn about Lambda Event Sources. We cover key concepts, practical examples, and best practices to help you master this topic.

AWS Lambda event sources are AWS services that invoke your function in response to events, enabling event-driven architectures where functions react to changes across your infrastructure.

What You'll Learn

By the end of this lesson you will understand the major Lambda event sources, how to configure triggers, handle event payloads from different sources, and choose the right source for your use case.

Why It Matters

Event sources are what make Lambda powerful. Instead of polling for changes, Lambda reacts instantly to S3 uploads, database changes, queue messages, and API requests -- reducing latency and eliminating polling costs.

Real-World Use

Doda Browser's user feedback system uses multiple event sources: API Gateway for HTTP submissions, S3 events for file attachments, DynamoDB Streams for analytics aggregation, and SQS for decoupling notification delivery.

flowchart TD
    AG[API Gateway HTTP] --> L[AWS Lambda]
    S3[S3 Upload] --> L
    DDB[DynamoDB Streams] --> L
    SQS[SQS Queue] --> L
    SNS[SNS Topic] --> L
    EB[EventBridge Schedule] --> L
    K[Kinesis Stream] --> L
    L --> P[Process Event]
    style L fill:#f90,color:#fff

Synchronous Event Sources

Synchronous sources like API Gateway and ALB expect an immediate response from Lambda. The function must return a response within the timeout or the request fails.

# sync_sources.py
# Handling synchronous event sources

import json

def handle_api_gateway(event, context):
    """API Gateway proxy integration."""
    method = event["httpMethod"]
    path = event["path"]
    body = json.loads(event.get("body", "null") or "null")
    
    if method == "GET" and path == "/items":
        items = [{"id": 1, "name": "Item 1"}]
        return {"statusCode": 200, "body": json.dumps(items)}
    
    if method == "POST" and path == "/items":
        new_item = {"id": 2, **body}
        return {"statusCode": 201, "body": json.dumps(new_item)}
    
    return {"statusCode": 404, "body": json.dumps({"error": "Not found"})}

def handle_alb(event, context):
    """Application Load Balancer target group."""
    method = event["httpMethod"]
    path = event["path"]
    return {
        "statusCode": 200,
        "statusDescription": "200 OK",
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"method": method, "path": path})
    }

api_event = {"httpMethod": "GET", "path": "/items", "body": None}
print(f"API Gateway: {handle_api_gateway(api_event, None)['body']}")

alb_event = {"httpMethod": "GET", "path": "/health"}
print(f"ALB: {handle_alb(alb_event, None)['body']}")

Expected output:

API Gateway: [{"id": 1, "name": "Item 1"}]
ALB: {"method": "GET", "path": "/health"}

Asynchronous Event Sources

Async sources like S3 and SNS invoke Lambda and do not wait for a response. Failed invocations are retried automatically.

# async_sources.py
# Handling async event sources

def handle_s3_event(event, context):
    """S3 event notification."""
    for record in event["Records"]:
        bucket = record["s3"]["bucket"]["name"]
        key = record["s3"]["object"]["key"]
        size = record["s3"]["object"]["size"]
        print(f"[S3] Object created: s3://{bucket}/{key} ({size} bytes)")

def handle_sns_event(event, context):
    """SNS topic subscription."""
    for record in event["Records"]:
        sns = record["Sns"]
        message = json.loads(sns["Message"])
        subject = sns.get("Subject", "No Subject")
        print(f"[SNS] Subject: {subject}")
        print(f"[SNS] Message: {message}")

def handle_eventbridge_event(event, context):
    """EventBridge scheduled event."""
    detail = event.get("detail", {})
    source = event.get("source", "unknown")
    time = event.get("time", "unknown")
    print(f"[EventBridge] {source} at {time}: {detail}")

s3_event = {"Records": [{"s3": {"bucket": {"name": "my-bucket"}, "object": {"key": "file.jpg", "size": 1024}}}]}
handle_s3_event(s3_event, None)

sns_event = {"Records": [{"Sns": {"Subject": "New Order", "Message": '{"order_id": "123"}'}}]}
handle_sns_event(sns_event, None)

eb_event = {"source": "aws.ec2", "time": "2026-06-28T00:00:00Z", "detail": {"state": "running"}}
handle_eventbridge_event(eb_event, None)

Expected output:

[S3] Object created: s3://my-bucket/file.jpg (1024 bytes)
[SNS] Subject: New Order
[SNS] Message: {'order_id': '123'}
[EventBridge] aws.ec2 at 2026-06-28T00:00:00Z: {'state': 'running'}

Stream-Based Event Sources

DynamoDB Streams and Kinesis invoke Lambda with batches of records. Lambda processes records in order within each shard and retries failed batches.

# stream_sources.py
# Handling stream-based event sources

def handle_dynamodb_stream(event, context):
    """DynamoDB Streams event."""
    for record in event["Records"]:
        event_name = record["eventName"]
        keys = record["dynamodb"]["Keys"]
        if "NewImage" in record["dynamodb"]:
            new_image = record["dynamodb"]["NewImage"]
            print(f"[DynamoDB] {event_name}: {keys} -> {new_image}")
        else:
            print(f"[DynamoDB] {event_name}: {keys} (deleted)")

def handle_kinesis_stream(event, context):
    """Kinesis stream event."""
    for record in event["Records"]:
        encoded_data = record["kinesis"]["data"]
        import base64
        data = json.loads(base64.b64decode(encoded_data))
        sequence = record["kinesis"]["sequenceNumber"]
        print(f"[Kinesis] Record {sequence}: {data}")

ddb_event = {"Records": [{"eventName": "INSERT", "dynamodb": {"Keys": {"id": {"S": "1"}}, "NewImage": {"id": {"S": "1"}, "name": {"S": "Alice"}}}}]}
handle_dynamodb_stream(ddb_event, None)

kinesis_event = {"Records": [{"kinesis": {"data": "eyJ1c2VyX2lkIjogIjEyMyJ9", "sequenceNumber": "123"}}]}
handle_kinesis_stream(kinesis_event, None)

Expected output:

[DynamoDB] INSERT: {'id': {'S': '1'}} -> {'id': {'S': '1'}, 'name': {'S': 'Alice'}}
[Kinesis] Record 123: {'user_id': '123'}

Queue-Based Event Sources

SQS invokes Lambda with batches of messages. Lambda deletes successfully processed messages from the queue. Failed messages return to the queue for retry or go to a dead-letter queue.

# sqs_source.py
# Handling SQS event source

def handle_sqs_event(event, context):
    """SQS event source."""
    for record in event["Records"]:
        message_id = record["messageId"]
        body = json.loads(record["body"])
        receipt_handle = record["receiptHandle"]
        
        print(f"[SQS] Processing message {message_id}: {body}")
        
        if body.get("type") == "process_image":
            print(f"  -> Image processing for {body['image_id']}")
        elif body.get("type") == "send_email":
            print(f"  -> Sending email to {body['email']}")
        else:
            print(f"  -> Unknown message type")

sqs_event = {"Records": [
    {"messageId": "1", "receiptHandle": "abc", "body": '{"type": "process_image", "image_id": "img_001"}'},
    {"messageId": "2", "receiptHandle": "def", "body": '{"type": "send_email", "email": "user@example.com"}'}
]}
handle_sqs_event(sqs_event, None)

Expected output:

[SQS] Processing message 1: {'type': 'process_image', 'image_id': 'img_001'}
  -> Image processing for img_001
[SQS] Processing message 2: {'type': 'send_email', 'email': 'user@example.com'}
  -> Sending email to user@example.com

Common Mistakes

Not handling partial batch failures: For SQS and Kinesis, use reportBatchItemFailures to retry only failed items instead of the entire batch.
Setting low reserved concurrency for stream sources: Stream sources need enough concurrency to keep up with the stream shard count. Insufficient concurrency causes throttling.
Ignoring event source mapping state: Event source mappings can be enabled, disabled, or in a failed state. Monitor them in the Lambda console.
Not configuring dead-letter queues: Unprocessed events are discarded without DLQ. Configure DLQs for all production event sources.
Using sync sources for long-running operations: API Gateway has a 29-second timeout. For operations taking longer, use SQS or Step Functions with async invocation.

Practice Questions

What is the difference between synchronous and asynchronous event sources? Synchronous sources wait for a response. Asynchronous sources invoke and forget, with automatic retries on failure.
How does Lambda handle SQS batch processing? Lambda polls the queue, invokes the function with up to 10 messages, and deletes successfully processed messages.
What happens when a Lambda function fails processing a DynamoDB Stream record? Lambda retries the entire batch until success or record expiry. The stream preserves record order within the shard.
Can a Lambda function have multiple event sources? Yes. A function can have multiple event source mappings, or be invoked directly by multiple services.
Challenge: Design an event-driven order processing system using SQS for order intake, DynamoDB Streams for analytics, and SNS for notifications.

FAQ

Can I trigger Lambda from my own application?

Yes. Use the AWS SDK Invoke API for direct invocation, or send events via SQS, SNS, or API Gateway.

How does Lambda scaling work with event sources?

Each event source has its own scaling behavior. SQS scales based on queue depth. Stream sources scale per shard.

What is the maximum SQS batch size?

10,000 records per batch with a payload limit of 6MB. Configure batch size and window in the event source mapping.

Can I filter events before they reach Lambda?

Yes. SQS, SNS, and EventBridge support event filtering using content-based filter policies.

How do I test event source configurations locally?

Use AWS SAM with local invoke, or the serverless-offline plugin for the Serverless Framework.

Mini Project

Create a Lambda function that handles three event sources: API Gateway for creating orders, SQS for processing orders asynchronously, and DynamoDB Streams for order analytics.

import json

def lambda_handler(event, context):
    if "httpMethod" in event:
        return handle_api(event, context)
    if "Records" in event and "messageId" in event["Records"][0]:
        return handle_sqs(event, context)
    if "Records" in event and "dynamodb" in event["Records"][0]:
        return handle_stream(event, context)
    return {"statusCode": 400, "body": json.dumps({"error": "Unknown source"})}

def handle_api(event, context):
    body = json.loads(event.get("body", "{}"))
    print(f"[API] Order created: {body}")
    return {"statusCode": 201, "body": json.dumps({"order_id": "ORD-001", "status": "created"})}

def handle_sqs(event, context):
    for record in event["Records"]:
        msg = json.loads(record["body"])
        print(f"[SQS] Processing order {msg['order_id']}: {msg['action']}")
    return {"batchItemFailures": []}

def handle_stream(event, context):
    for record in event["Records"]:
        image = record["dynamodb"].get("NewImage", {})
        print(f"[Stream] Order event: {image}")

# Test all three sources
api_event = {"httpMethod": "POST", "path": "/orders", "body": '{"product": "book", "qty": 1}'}
sqs_event = {"Records": [{"messageId": "1", "receiptHandle": "h", "body": '{"order_id": "1", "action": "process"}'}]}
stream_event = {"Records": [{"eventName": "INSERT", "dynamodb": {"Keys": {}, "NewImage": {"status": {"S": "completed"}}}}]}

print(lambda_handler(api_event, None)["body"])
lambda_handler(sqs_event, None)
lambda_handler(stream_event, None)

What's Next

Next: Lambda + API Gateway to build RESTful APIs with Serverless.

← Previous Lambda Environment Variables — Configuration Management Next → Lambda + API Gateway — Building Serverless REST APIs

Built by the developers of DodaTech

Doda Browser, DodaZIP & Durga Antivirus Pro

Home Browse Serverless