Lambda Functions — Writing and Structuring Code
In this tutorial, you will learn about Lambda Functions. We cover key concepts, practical examples, and best practices to help you master this topic.
AWS Lambda functions follow a specific handler pattern where your code receives an event and context object and returns a response, with the execution environment reused across warm invocations.
What You'll Learn
By the end of this lesson you will know how to structure Lambda function code, handle different event sources, implement proper error handling, and optimize for cold start performance.
Why It Matters
Poorly structured Lambda functions are hard to debug, expensive to run, and fail silently. Following consistent patterns ensures your functions are maintainable, testable, and cost-effective at scale.
Real-World Use
DodaZIP's file conversion service uses Lambda functions with a clean handler-service-repository structure. The handler parses the event, the service layer contains business logic, and the repository layer handles data access -- making each layer independently testable.
flowchart TD
H[Handler: parse event] --> S[Service: business logic]
S --> R[Repository: data access]
R --> D[DynamoDB]
S --> E[External APIs]
H --> L[Logging]
H --> E1[Error Handler]
style H fill:#f90,color:#fff
Handler Structure
The handler function is the entry point. Keep it thin -- parse the event, call business logic, and return a response. The actual logic belongs in separate modules.
# lambda_structure.py
# Well-structured Lambda function
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Service layer
class UserService:
def __init__(self, db_client):
self.db = db_client
def get_user(self, user_id):
logger.info(f"Fetching user {user_id}")
if not user_id:
raise ValueError("user_id is required")
return {"id": user_id, "name": "Alice", "email": "alice@example.com"}
# Handler layer
def lambda_handler(event, context):
try:
path_params = event.get("pathParameters", {}) or {}
user_id = path_params.get("user_id")
service = UserService(db_client=None)
user = service.get_user(user_id)
return {
"statusCode": 200,
"headers": {"Content-Type": "application/json"},
"body": json.dumps(user)
}
except ValueError as e:
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except Exception as e:
logger.exception("Unexpected error")
return {"statusCode": 500, "body": json.dumps({"error": "Internal server error"})}
Event Parsing Patterns
Different event sources produce different event shapes. API Gateway sends HTTP-like events. S3 sends bucket and key information. DynamoDB Streams send record changes.
# event_parsers.py
# Parsing events from different sources
def parse_api_gateway_event(event):
"""Extract HTTP request details from API Gateway event."""
method = event.get("httpMethod", "GET")
path = event.get("path", "/")
body = json.loads(event.get("body", "null") or "null")
params = event.get("queryStringParameters", {}) or {}
headers = event.get("headers", {})
return {"method": method, "path": path, "body": body, "params": params, "headers": headers}
def parse_s3_event(event):
"""Extract S3 bucket and key from S3 event."""
records = event.get("Records", [])
events = []
for record in records:
bucket = record["s3"]["bucket"]["name"]
key = record["s3"]["object"]["key"]
events.append({"bucket": bucket, "key": key})
return events
def parse_sqs_event(event):
"""Extract messages from SQS event."""
messages = []
for record in event.get("Records", []):
body = json.loads(record["body"])
message_id = record["messageId"]
messages.append({"id": message_id, "body": body})
return messages
api_event = {"httpMethod": "GET", "path": "/users/1"}
print(f"API: {parse_api_gateway_event(api_event)}")
s3_event = {"Records": [{"s3": {"bucket": {"name": "my-bucket"}, "object": {"key": "uploads/image.jpg"}}}]}
print(f"S3: {parse_s3_event(s3_event)}")
sqs_event = {"Records": [{"messageId": "msg1", "body": '{"type": "order.placed"}'}]}
print(f"SQS: {parse_sqs_event(sqs_event)}")
Expected output:
API: {'method': 'GET', 'path': '/users/1', 'body': None, 'params': {}, 'headers': {}}
S3: [{'bucket': 'my-bucket', 'key': 'uploads/image.jpg'}]
SQS: [{'id': 'msg1', 'body': {'type': 'order.placed'}}]
Error Handling
Distinguish between client errors (4xx), server errors (5xx), and transient errors (retries). Use custom exceptions for domain errors.
# error_handling.py
# Lambda error handling patterns
class NotFoundError(Exception):
pass
class ValidationError(Exception):
pass
def lambda_handler(event, context):
try:
user_id = event.get("pathParameters", {}).get("id")
if not user_id:
raise ValidationError("Missing user ID")
user = find_user(user_id)
if not user:
raise NotFoundError(f"User {user_id} not found")
return {"statusCode": 200, "body": json.dumps(user)}
except ValidationError as e:
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except NotFoundError as e:
return {"statusCode": 404, "body": json.dumps({"error": str(e)})}
def find_user(user_id):
return None if user_id == "999" else {"id": user_id, "name": "Alice"}
print(lambda_handler({"pathParameters": {}}, None))
print(lambda_handler({"pathParameters": {"id": "999"}}, None))
print(lambda_handler({"pathParameters": {"id": "1"}}, None))
Expected output:
{'statusCode': 400, 'body': '{"error": "Missing user ID"}'}
{'statusCode': 404, 'body': '{"error": "User 999 not found"}'}
{'statusCode': 200, 'body': '{"id": "1", "name": "Alice"}'}
Logging Best Practices
Use structured JSON logging for CloudWatch Logs Insights queries. Include request IDs and correlation IDs for tracing.
import json
import logging
import os
class StructuredLogger:
def __init__(self, service_name):
self.service = service_name
def info(self, message, **kwargs):
log_entry = {"level": "INFO", "service": self.service, "message": message, **kwargs}
print(json.dumps(log_entry))
def error(self, message, **kwargs):
log_entry = {"level": "ERROR", "service": self.service, "message": message, **kwargs}
print(json.dumps(log_entry))
logger = StructuredLogger("user-service")
logger.info("User created", user_id="123", source="signup")
logger.error("Database timeout", table="users-table", duration_ms=5000)
Expected output:
{"level": "INFO", "service": "user-service", "message": "User created", "user_id": "123", "source": "signup"}
{"level": "ERROR", "service": "user-service", "message": "Database timeout", "table": "users-table", "duration_ms": 5000}
Common Mistakes
Putting too much logic in the handler: Handlers should parse and delegate. Business logic belongs in separate, testable modules.
Not handling partial failures: When processing batch events from SQS or Kinesis, report partial failures so unprocessed items are retried.
Forgetting to return a response: API Gateway expects a specific response format with statusCode and body.
Logging sensitive information: Never log passwords, tokens, or personal data. Use structured logging with redaction.
Relying on /tmp for persistence: The /tmp directory exists only for the lifecycle of the execution environment. Use S3 or EFS for durable storage.
Practice Questions
What is the purpose of the context object in Lambda? It provides runtime information like request ID, function name, timeout, and identity details for the current invocation.
How should you structure a Lambda function with multiple responsibilities? Use separate modules for handler, service, and data access layers. Keep the handler thin.
What happens when a Lambda function throws an unhandled exception? The invocation fails. For synchronous invocations the caller receives an error. For async invocations Lambda retries twice.
How do you handle different event sources in one function? Check the event structure at the handler level and route to the appropriate internal function.
Challenge: Write a Lambda function that handles both API Gateway HTTP requests and S3 events, routing to different handlers based on the event source.
FAQ
Mini Project
Create a Lambda function with three layers: a handler that parses the event, a service layer that processes image metadata, and a repository layer that stores results in DynamoDB.
import json
import uuid
class MetadataRepository:
def save(self, metadata):
metadata_id = str(uuid.uuid4())
print(f"[Repository] Saved metadata {metadata_id}: {json.dumps(metadata)}")
return metadata_id
class ImageService:
def __init__(self, repo):
self.repo = repo
def process_image(self, bucket, key):
print(f"[Service] Processing s3://{bucket}/{key}")
metadata = {"bucket": bucket, "key": key, "size_bytes": 1024000, "format": "JPEG"}
metadata_id = self.repo.save(metadata)
return {"id": metadata_id, "metadata": metadata}
def lambda_handler(event, context):
records = event.get("Records", [])
results = []
repo = MetadataRepository()
service = ImageService(repo)
for record in records:
bucket = record["s3"]["bucket"]["name"]
key = record["s3"]["object"]["key"]
result = service.process_image(bucket, key)
results.append(result)
return {"statusCode": 200, "body": json.dumps(results)}
test_event = {"Records": [{"s3": {"bucket": {"name": "uploads"}, "object": {"key": "photos/sunset.jpg"}}}]}
print(lambda_handler(test_event, None)["body"])
What's Next
Next: Lambda Layers to manage shared dependencies across multiple functions.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro