Lambda + DynamoDB — Serverless Database Patterns
In this tutorial, you will learn about Lambda + DynamoDB. We cover key concepts, practical examples, and best practices to help you master this topic.
AWS Lambda combined with DynamoDB provides a fully managed, scalable Serverless database pattern that handles CRUD operations, real-time streams, and complex queries without provisioning servers.
What You'll Learn
By the end of this lesson you will understand how to perform CRUD operations from Lambda, use DynamoDB Streams for real-time processing, implement single-table design, and optimize read/write performance.
Why It Matters
DynamoDB is the most common database for serverless applications. It scales automatically, has single-digit millisecond latency, and integrates natively with Lambda through Streams -- making it the default choice for event-driven serverless architectures.
Real-World Use
Doda Browser's session management system stores user sessions in DynamoDB. Lambda functions handle session creation at login, session validation per request, and session cleanup on expiry -- with DynamoDB Streams triggering analytics for active user counts.
flowchart LR
AG[API Gateway] --> L[AWS Lambda]
L --> D[DynamoDB Table]
L --> S[DynamoDB Streams]
S --> L2[Analytics Lambda]
S --> L3[Cleanup Lambda]
style D fill:#f90,color:#fff
Basic CRUD Operations
Lambda functions interact with DynamoDB using the AWS SDK. Each operation requires the table name and appropriate IAM permissions.
# dynamodb_crud.py
# CRUD operations from Lambda
import json
import boto3
from boto3.dynamodb.conditions import Key
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table("users")
def lambda_handler(event, context):
method = event.get("httpMethod", "GET")
path = event.get("path", "/")
if method == "POST":
body = json.loads(event["body"])
response = table.put_item(
Item={
"userId": body["userId"],
"name": body["name"],
"email": body["email"],
"active": True
}
)
return {"statusCode": 201, "body": json.dumps({"userId": body["userId"]})}
if method == "GET" and path == "/users":
response = table.scan()
return {"statusCode": 200, "body": json.dumps(response.get("Items", []))}
if method == "GET":
user_id = event["pathParameters"]["userId"]
response = table.get_item(Key={"userId": user_id})
item = response.get("Item")
if not item:
return {"statusCode": 404, "body": json.dumps({"error": "Not found"})}
return {"statusCode": 200, "body": json.dumps(item)}
if method == "PUT":
user_id = event["pathParameters"]["userId"]
body = json.loads(event["body"])
response = table.update_item(
Key={"userId": user_id},
UpdateExpression="SET #n = :name, email = :email",
ExpressionAttributeNames={"#n": "name"},
ExpressionAttributeValues={":name": body["name"], ":email": body["email"]}
)
return {"statusCode": 200, "body": json.dumps({"updated": user_id})}
if method == "DELETE":
user_id = event["pathParameters"]["userId"]
table.delete_item(Key={"userId": user_id})
return {"statusCode": 204, "body": ""}
# Simulate (using dict as mock table)
print("CRUD patterns demonstrated with boto3 DynamoDB resource.")
print("Operations: put_item, get_item, update_item, delete_item, scan")
Query vs Scan
Query is efficient and uses the primary key. Scan reads the entire table and should be avoided in production. Use Global Secondary Indexes for alternative query patterns.
# query_vs_scan.py
# Query vs Scan performance comparison
def simulate_query(records, partition_key, sort_key_prefix):
items = [r for r in records if r["pk"] == partition_key and r["sk"].startswith(sort_key_prefix)]
print(f"Query: Read 1 partition, returned {len(items)} items")
return items
def simulate_scan(records, filter_field, filter_value):
items = [r for r in records if r.get(filter_field) == filter_value]
print(f"Scan: Read ALL {len(records)} records, filtered to {len(items)} items")
return items
records = [{"pk": f"USER#{i}", "sk": f"ORDER#{j}", "status": "active"} for i in range(10) for j in range(100)]
query_result = simulate_query(records, "USER#1", "ORDER#")
scan_result = simulate_scan(records, "status", "active")
print(f"\nQuery cost: {len([r for r in records if r['pk'] == 'USER#1'])} RCUs")
print(f"Scan cost: {len(records)} RCUs (entire table read)")
Expected output:
Query: Read 1 partition, returned 100 items
Scan: Read ALL 1000 records, filtered to 1000 items
Query cost: 100 RCUs
Scan cost: 1000 RCUs (entire table read)
DynamoDB Streams Trigger
DynamoDB Streams capture changes to table items and invoke Lambda functions with before-and-after images of modified items.
# streams_trigger.py
# DynamoDB Streams Lambda trigger
import json
def lambda_handler(event, context):
for record in event["Records"]:
event_name = record["eventName"]
keys = record["dynamodb"]["Keys"]
approximate_time = record["dynamodb"]["ApproximateCreationDateTime"]
old_image = record["dynamodb"].get("OldImage", {})
new_image = record["dynamodb"].get("NewImage", {})
print(f"[{event_name}] {keys} at {approximate_time}")
if event_name == "INSERT":
handle_insert(new_image)
elif event_name == "MODIFY":
handle_modify(old_image, new_image)
elif event_name == "REMOVE":
handle_remove(old_image)
def handle_insert(new_image):
print(f" New item created: {new_image}")
def handle_modify(old_image, new_image):
changed_fields = {k: {"from": old_image.get(k), "to": new_image.get(k)}
for k in set(list(old_image.keys()) + list(new_image.keys()))
if old_image.get(k) != new_image.get(k)}
print(f" Changed fields: {changed_fields}")
def handle_remove(old_image):
print(f" Item deleted: {old_image}")
mock_event = {"Records": [
{"eventName": "INSERT", "dynamodb": {"Keys": {"id": {"S": "1"}}, "NewImage": {"id": {"S": "1"}, "status": {"S": "active"}}, "ApproximateCreationDateTime": 1719500000}},
{"eventName": "MODIFY", "dynamodb": {"OldImage": {"status": {"S": "active"}}, "NewImage": {"status": {"S": "inactive"}}, "Keys": {"id": {"S": "1"}}, "ApproximateCreationDateTime": 1719500001}}
]}
lambda_handler(mock_event, None)
Expected output:
[INSERT] {'id': {'S': '1'}} at 1719500000
New item created: {'id': {'S': '1'}, 'status': {'S': 'active'}}
[MODIFY] {'id': {'S': '1'}} at 1719500001
Changed fields: {'status': {'from': {'S': 'active'}, 'to': {'S': 'inactive'}}}
Single-Table Design
DynamoDB works best with a single table design where different entity types share the same table using Composite keys.
# single_table.py
# Single-table design pattern
def create_user_record(user_id, name, email):
return {
"pk": f"USER#{user_id}",
"sk": "PROFILE",
"entity_type": "USER",
"name": name,
"email": email,
"created_at": "2026-06-28"
}
def create_order_record(order_id, user_id, amount):
return {
"pk": f"USER#{user_id}",
"sk": f"ORDER#{order_id}",
"entity_type": "ORDER",
"amount": amount,
"status": "pending",
"created_at": "2026-06-28"
}
def query_user_orders(records, user_id):
return [r for r in records if r["pk"] == f"USER#{user_id}" and r["entity_type"] == "ORDER"]
table = []
table.append(create_user_record("1", "Alice", "alice@example.com"))
table.append(create_order_record("ORD-001", "1", 49.99))
table.append(create_order_record("ORD-002", "1", 29.99))
table.append(create_user_record("2", "Bob", "bob@example.com"))
orders = query_user_orders(table, "1")
print(f"Alice has {len(orders)} orders:")
for order in orders:
print(f" {order['sk']}: ${order['amount']}")
Expected output:
Alice has 2 orders:
ORDER#ORD-001: $49.99
ORDER#ORD-002: $29.99
Common Mistakes
Using Scan instead of Query: Scan reads every item in the table. Always design your schema so queries use partition keys.
Not setting appropriate RCU/WCU: Provisioned capacity can throttle under load. Use on-demand capacity for variable workloads.
Ignoring item size limits: DynamoDB items are limited to 400KB. Store large objects in S3 with a reference in DynamoDB.
Forgetting TTL for ephemeral data: Enable TTL on tables with session data or event logs to automatically delete expired items.
Missing error handling for ConditionalCheckFailedException: Conditional writes can fail when conditions are not met. Handle this exception gracefully.
Practice Questions
What is the difference between Query and Scan in DynamoDB? Query reads items by partition key and optional sort key. Scan reads the entire table sequentially.
How do DynamoDB Streams work with Lambda? Streams capture item-level changes. Lambda polls the stream and invokes the function with batches of change records.
What is the maximum item size in DynamoDB? 400KB including attribute names. Store larger payloads in S3 with a DynamoDB reference.
What is single-table design? Storing multiple entity types in one table using composite primary keys (pk and sk), enabling complex query patterns.
Challenge: Design a DynamoDB table and Lambda functions for a task management system with users, projects, and tasks using single-table design.
FAQ
Mini Project
Create Lambda functions for a DynamoDB-backed URL shortener: store short code to URL mapping, handle redirects, track click counts, and use Streams for analytics.
import json
import hashlib
import time
urls = {}
def lambda_handler(event, context):
method = event["httpMethod"]
path = event.get("path", "/")
if method == "POST":
body = json.loads(event["body"])
long_url = body["url"]
code = hashlib.md5(long_url.encode()).hexdigest()[:8]
urls[code] = {"url": long_url, "clicks": 0, "created": time.time()}
return {"statusCode": 201, "body": json.dumps({"short_url": f"https://short.ly/{code}"})}
if method == "GET" and path.startswith("/"):
code = path.strip("/")
if code in urls:
urls[code]["clicks"] += 1
return {"statusCode": 301, "headers": {"Location": urls[code]["url"]}, "body": ""}
return {"statusCode": 404, "body": json.dumps({"error": "Not found"})}
return {"statusCode": 400, "body": json.dumps({"error": "Bad request"})}
print(lambda_handler({"httpMethod": "POST", "body": json.dumps({"url": "https://example.com/long/url"})}, None)["body"])
code = "a1b2c3d4"
urls[code] = {"url": "https://example.com", "clicks": 0, "created": time.time()}
print(lambda_handler({"httpMethod": "GET", "path": f"/{code}"}, None))
What's Next
Next: Lambda + S3 for file processing patterns.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro