THE COMPLETE GUIDE TO
AWS LAMBDA
Python Edition
l
Functions Events Layers VPC IAM API GW SAM CDK Monitoring Best Practices
Python 3.12 | boto3 | AWS SDK | SAM | CDK | 2025 Edition
Serverless Architecture • Code Examples • Diagrams • Best Practices
AWS Lambda Complete Guide | Python Edition | 2025 Page 1
AWS Lambda Complete Guide | Python Edition | 2025 Page 1
AWS TABLE OF CONTENTS AWS Lambda — Python Complete Guide
Table of Contents
1. What is AWS Lambda? — Serverless Fundamentals . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
2. Core Concepts — Functions, Events, Context, Runtimes . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
3. Your First Lambda — Hello World in Python . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
4. Execution Lifecycle — Cold Starts, Warm Reuse, Init Phase . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
5. Event Sources & Triggers — API GW, S3, SQS, DynamoDB... . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
6. IAM Roles & Permissions — Execution Role, Resource Policies . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
7. Environment Variables & Configuration . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
8. Lambda Layers — Shared Code & Dependencies . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
9. Working with S3 — Read, Write, Trigger . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
10. Working with DynamoDB — CRUD with boto3 . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
11. Working with SQS & SNS — Queues and Notifications . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
12. API Gateway + Lambda — Building REST APIs . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
13. Lambda inside a VPC — RDS, ElastiCache, Private Resources . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
14. Error Handling, Retries & Dead Letter Queues . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
15. Lambda Destinations — Async Success/Failure Routing . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
16. Observability — CloudWatch, X-Ray, Structured Logging . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
17. Performance — Memory, Timeout, Concurrency, Provisioned . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
18. Lambda Extensions & SnapStart . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
19. Deploying with SAM — Template, Build, Local Test, Deploy . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
20. Deploying with AWS CDK — Python CDK Constructs . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
21. CI/CD for Lambda — GitHub Actions Pipeline . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
22. Security Best Practices . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
23. Cost Optimisation & Limits Reference . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
24. Complete Cheat Sheet & Quick Reference . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . .
AWS Lambda Complete Guide | Python Edition | 2025 Page 2
AWS 1 — WHAT IS AWS LAMBDA? AWS Lambda — Python Complete Guide
1 What is AWS Lambda?
Serverless Computing
AWS Lambda is a serverless compute service that lets you run code without provisioning or managing servers. You upload your
code (or a container image), configure a trigger, and Lambda handles everything else — scaling, patching, availability, and
billing. You pay only for the compute time your code actually consumes, measured in milliseconds.
Key Benefits
• No servers to provision, patch, or manage — zero infrastructure overhead.
• Automatic scaling from zero to thousands of concurrent executions in seconds.
• Pay-per-use pricing: charged for requests + duration (GB-seconds), not idle time.
• Built-in high availability across multiple Availability Zones.
• Native integration with 200+ AWS services as event sources or destinations.
• Supports Python, [Link], Java, Go, Ruby, .NET, and custom runtimes.
• Run code in response to virtually any event: HTTP requests, file uploads, DB changes, schedules.
When to Use Lambda (and When Not To)
Good Fit Poor Fit
Event-driven, short-lived tasks (< 15 min) Long-running jobs (> 15 minutes)
Unpredictable or spiky traffic patterns Steady high-throughput workloads (EC2 cheaper)
Microservices and API backends Stateful applications needing persistent connections
File processing, data transformation Real-time streaming requiring sub-ms latency
Scheduled tasks (cron-style) Large binaries or dependencies > 250 MB unzipped
Webhook handlers, notifications Workloads needing GPU or specialised hardware
Lambda vs EC2 vs Fargate
Aspect Lambda EC2 Fargate
Management None Full (OS, patches) Container only
Scaling Automatic (per-request) Manual / ASG Task-level auto scaling
Billing Per request + ms Per hour (running) Per vCPU + memory-second
Max duration 15 minutes Unlimited Unlimited
State Stateless Stateful Stateless by default
Cold start Yes (ms to s) N/A (always on) Yes (slower than Lambda)
Best for Event-driven, short tasks Persistent workloads Containerised services
AWS Lambda Complete Guide | Python Edition | 2025 Page 3
AWS 2 — CORE CONCEPTS AWS Lambda — Python Complete Guide
2 Core Concepts
Anatomy of a Lambda Function
Component Description
Function code Your Python module (or container image) with the handler ent
Handler The Python function Lambda calls: module_name.function_name
Event JSON-serialisable dict passed to your handler — shape depend
Context Runtime information object: function name, memory, request I
Execution role IAM role granting Lambda permission to call other AWS servic
Memory 128 MB – 10,240 MB. CPU is allocated proportionally to memor
Timeout 1 s – 900 s (15 minutes). Lambda raises a Timeout error if e
Ephemeral storage /tmp up to 10,240 MB for temporary files within an invocatio
Runtime Managed Python 3.8 / 3.9 / 3.10 / 3.11 / 3.12, or custom run
The handler Signature
python
import json
def handler(event: dict, context) -> dict:
"""
event — dict deserialized from the trigger's JSON payload
context — LambdaContext object with runtime metadata
"""
# context attributes
print(context.function_name) # e.g. 'my-function'
print(context.function_version) # e.g. '$LATEST' or '3'
print(context.memory_limit_in_mb) # e.g. '512'
print(context.aws_request_id) # unique per invocation
print(context.invoked_function_arn) # full ARN
ms_remaining = context.get_remaining_time_in_millis()
# Your logic here
name = [Link]('name', 'World')
return {
'statusCode': 200,
'body': [Link]({'message': f'Hello, {name}!'})
}
Invocation Types
Type Behaviour Example Triggers
Synchronous (RequestResponse) Caller waits for response; errors returned directly API Gateway, ALB, CLI invoke
Asynchronous (Event) Lambda queues the event; caller gets 202 immediately; S3,
2 retr
SNS, EventBridge, SES
Poll-based (Stream/Queue) Lambda polls the source; batches records for your handler
SQS, Kinesis, DynamoDB Streams, MSK
Supported Python Runtimes (2025)
Runtime ID Python Version Status Notes
python3.12 3.12.x Current (recommended) Best performance, latest stdlib
python3.11 3.11.x Supported Good choice for mature projects
python3.10 3.10.x Supported Match patterns, parenthesised context managers
python3.9 3.9.x Supported End-of-support approaching
python3.8 3.8.x Deprecated Migrate to 3.12 recommended
AWS Lambda Complete Guide | Python Edition | 2025 Page 4
AWS 3 — YOUR FIRST LAMBDA FUNCTION AWS Lambda — Python Complete Guide
3 Your First Lambda — Python
Project Structure
text
my-lambda/
lambda_function.py # handler lives here
[Link] # pip dependencies
tests/
test_handler.py # unit tests
[Link] # SAM template (optional)
Hello World Handler
python
# lambda_function.py
import json
import logging
logger = [Link]()
[Link]([Link])
def handler(event: dict, context) -> dict:
"""Minimal Lambda handler — returns a greeting."""
[Link]('Event received: %s', [Link](event))
name = [Link]('queryStringParameters', {}) or {}
name = [Link]('name', [Link]('name', 'World'))
response_body = {'message': f'Hello, {name}!', 'input': event}
return {
'statusCode': 200,
'headers': {'Content-Type': 'application/json'},
'body': [Link](response_body),
}
Deploy via AWS Console (quick start)
• Open AWS Console → Lambda → Create function
• Choose "Author from scratch", Runtime: Python 3.12
• Paste code into the inline editor → Deploy
• Click "Test", create a test event → Invoke
Deploy via AWS CLI
bash
# 1. Zip your code
zip [Link] lambda_function.py
# 2. Create function
aws lambda create-function \
--function-name HelloWorld \
--runtime python3.12 \
--role arn:aws:iam::123456789:role/lambda-execution-role \
--handler lambda_function.handler \
--zip-file fileb://[Link]
# 3. Invoke synchronously
aws lambda invoke \
--function-name HelloWorld \
--payload '{"name": "Alice"}' \
AWS Lambda Complete Guide | Python Edition | 2025 Page 5
AWS 3 — YOUR FIRST LAMBDA FUNCTION AWS Lambda — Python Complete Guide
--cli-binary-format raw-in-base64-out \
[Link]
cat [Link]
# 4. Update code after changes
zip [Link] lambda_function.py
aws lambda update-function-code \
--function-name HelloWorld \
--zip-file fileb://[Link]
# 5. Update configuration
aws lambda update-function-configuration \
--function-name HelloWorld \
--timeout 30 \
--memory-size 512
Unit Testing Locally
python
# tests/test_handler.py
import json
import unittest
from [Link] import MagicMock
from lambda_function import handler
class TestHandler([Link]):
def _make_context(self):
ctx = MagicMock()
ctx.function_name = 'test-function'
ctx.memory_limit_in_mb = '128'
ctx.aws_request_id = 'test-request-id'
ctx.get_remaining_time_in_millis.return_value = 30000
return ctx
def test_hello_world(self):
event = {'name': 'Alice'}
result = handler(event, self._make_context())
[Link](result['statusCode'], 200)
body = [Link](result['body'])
[Link]('Hello, Alice', body['message'])
def test_default_name(self):
result = handler({}, self._make_context())
body = [Link](result['body'])
[Link]('Hello, World', body['message'])
if __name__ == '__main__':
[Link]()
AWS Lambda Complete Guide | Python Edition | 2025 Page 6
AWS 4 — EXECUTION LIFECYCLE AWS Lambda — Python Complete Guide
4 Execution Lifecycle & Cold Starts
Lifecycle Diagram
Lambda Execution Lifecycle
Cold Init Invoke Warm Freeze /
Start (Execute) Invoke Teardown
warm reuse (no Init)
Phases Explained
Phase What Happens Your Code Runs?
Cold Start AWS provisions a new execution environment (download code, s No
Init Module-level code runs: imports, connections, global vars in Yes (module level)
Invoke Lambda calls your handler function with event + context Yes (handler)
Warm Reuse Same execution environment handles next request — Init skipp Yes (handler only)
Freeze Environment frozen after invocation completes, awaiting next No
Teardown Environment destroyed after idle period or deployment update No
Optimising for Cold Starts
python
# BAD: expensive setup inside handler (repeated every invocation)
def handler(event, context):
import boto3 # import on every call
s3 = [Link]('s3') # new client every call
db = connect_to_database() # new connection every call
...
# GOOD: module-level init (runs once during Init phase, reused)
import boto3
import logging
from database import connect_to_database # import at module level
logger = [Link]()
[Link]([Link])
# These run once during INIT, then reused across warm invocations
s3_client = [Link]('s3')
db_connection = connect_to_database()
def handler(event, context):
# s3_client and db_connection already exist — reuse them
result = db_connection.query('SELECT ...')
...
# GOOD: lazy init with module-level variable (safest pattern)
_db = None
def get_db():
global _db
if _db is None:
_db = connect_to_database()
return _db
def handler(event, context):
AWS Lambda Complete Guide | Python Edition | 2025 Page 7
AWS 4 — EXECUTION LIFECYCLE AWS Lambda — Python Complete Guide
db = get_db() # creates once, reuses on warm invocations
...
Cold Start Reduction Strategies
• Use Python 3.12 — it has the fastest Lambda cold start of all managed runtimes.
• Keep your deployment package small — smaller zip = faster code download.
• Move imports and SDK client init to module level (outside handler).
• Use Provisioned Concurrency for latency-sensitive APIs (pre-warms environments).
• Use Lambda SnapStart (available for Java; watch for Python support).
• Avoid large dependency trees — use Lambda Layers for shared libs.
• Consider keeping functions "warm" with EventBridge scheduled pings (workaround).
AWS Lambda Complete Guide | Python Edition | 2025 Page 8
AWS 5 — EVENT SOURCES & TRIGGERS AWS Lambda — Python Complete Guide
5 Event Sources & Triggers
Event Sources Overview
Lambda Event Sources (Triggers)
API Gateway EventBridge
/ ALB / Schedule
S3 Events SNS / SQS
AWS Lambda
handler(event, context)
DynamoDB Cognito /
Streams Kinesis
Event Source Mapping — All Major Triggers
Source Invocation Type Batch Support Notes
API Gateway (REST/HTTP) Sync No Returns response directly to HTTP client
Application Load Balancer Sync No Similar to API GW; supports multi-value headers
S3 (ObjectCreated, etc.) Async No Up to 3 retries; use DLQ for failures
SQS Poll-based Yes (1-10000) Lambda deletes messages on success; partial batch response s
SNS Async No Fan-out pattern; Lambda subscribed as topic subscriber
DynamoDB Streams Poll-based Yes In-order per partition; 4 retry attempts
Kinesis Data Streams Poll-based Yes In-order per shard; configurable bisect-on-error
EventBridge (scheduled) Async No Cron or rate expressions for scheduled tasks
Cognito (User Pools) Sync No Triggers: pre-signup, post-confirm, pre-token, etc.
CloudFront (Lambda@Edge) Sync No Runs at edge locations; < 5 MB, < 30s timeout
Reading Common Event Shapes
python
# ■■ API Gateway v2 (HTTP API) event ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def api_handler(event, context):
method = event['requestContext']['http']['method']
path = event['rawPath'] # e.g. '/users/42'
query = [Link]('queryStringParameters') or {}
headers = [Link]('headers') or {}
body_str = [Link]('body', '')
is_b64 = [Link]('isBase64Encoded', False)
if is_b64:
import base64
body_str = base64.b64decode(body_str).decode('utf-8')
import json
body = [Link](body_str) if body_str else {}
return {'statusCode': 200, 'body': [Link]({'ok': True})}
# ■■ S3 event ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def s3_handler(event, context):
for record in event['Records']:
bucket = record['s3']['bucket']['name']
AWS Lambda Complete Guide | Python Edition | 2025 Page 9
AWS 5 — EVENT SOURCES & TRIGGERS AWS Lambda — Python Complete Guide
key = record['s3']['object']['key']
size = record['s3']['object'].get('size', 0)
print(f'New object: s3://{bucket}/{key} ({size} bytes)')
# ■■ SQS event ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
import json
def sqs_handler(event, context):
failed_ids = []
for record in event['Records']:
msg_id = record['messageId']
body = [Link](record['body'])
try:
process_message(body)
except Exception as e:
print(f'Failed {msg_id}: {e}')
failed_ids.append({'itemIdentifier': msg_id})
# Partial batch response — only failed messages return to queue
return {'batchItemFailures': [{'itemIdentifier': i} for i in failed_ids]}
# ■■ EventBridge scheduled event ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def schedule_handler(event, context):
# event = {'version': '0', 'source': '[Link]', ...}
print(f'Scheduled run at: {event["time"]}')
run_daily_job()
AWS Lambda Complete Guide | Python Edition | 2025 Page 10
AWS 6 — IAM ROLES & PERMISSIONS AWS Lambda — Python Complete Guide
6 IAM Roles & Permissions
Execution Role
Every Lambda function must have an IAM execution role. Lambda assumes this role when your function runs. It defines what
AWS services and resources your code can access. The role must trust the [Link] service principal.
json
# Minimal execution role trust policy
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": { "Service": "[Link]" },
"Action": "sts:AssumeRole"
}]
}
# Always start with AWSLambdaBasicExecutionRole (CloudWatch Logs)
# Then add only the permissions your function needs
Example IAM Policy — S3 Read + DynamoDB Write
json
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "ReadSourceBucket",
"Effect": "Allow",
"Action": ["s3:GetObject", "s3:ListBucket"],
"Resource": [
"arn:aws:s3:::my-source-bucket",
"arn:aws:s3:::my-source-bucket/*"
]
},
{
"Sid": "WriteDynamoDB",
"Effect": "Allow",
"Action": ["dynamodb:PutItem", "dynamodb:UpdateItem"],
"Resource": "arn:aws:dynamodb:us-east-1:123456789:table/MyTable"
},
{
"Sid": "WriteLogs",
"Effect": "Allow",
"Action": ["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],
"Resource": "arn:aws:logs:*:*:*"
}
]
}
Resource-Based Policy (who can invoke Lambda)
bash
# Allow API Gateway to invoke your function
aws lambda add-permission \
--function-name MyFunction \
--statement-id allow-api-gateway \
--action lambda:InvokeFunction \
--principal [Link] \
--source-arn "arn:aws:execute-api:us-east-1:123456789:abc123/*/GET/users"
# Allow S3 to invoke (for event notifications)
AWS Lambda Complete Guide | Python Edition | 2025 Page 11
AWS 6 — IAM ROLES & PERMISSIONS AWS Lambda — Python Complete Guide
aws lambda add-permission \
--function-name MyFunction \
--statement-id allow-s3 \
--action lambda:InvokeFunction \
--principal [Link] \
--source-account 123456789 \
--source-arn arn:aws:s3:::my-bucket
# View current resource policy
aws lambda get-policy --function-name MyFunction
Least Privilege Best Practices
• Never attach AdministratorAccess or AmazonS3FullAccess — scope to specific actions.
• Restrict Resource to the exact ARN(s) your function needs, not "*".
• Use separate execution roles per function — avoid one shared role for all Lambdas.
• Use aws:SourceArn condition on resource policies to prevent confused-deputy attacks.
• Audit permissions with IAM Access Analyzer and AWS CloudTrail.
• Store sensitive values in Secrets Manager or Parameter Store — not IAM policies.
AWS Lambda Complete Guide | Python Edition | 2025 Page 12
AWS 7 — ENVIRONMENT VARIABLES & CONFIGURATION AWS Lambda — Python Complete Guide
7 Environment Variables & Configuration
Setting Environment Variables
bash
# Via AWS CLI
aws lambda update-function-configuration \
--function-name MyFunction \
--environment 'Variables={
TABLE_NAME=my-table,
BUCKET_NAME=my-bucket,
LOG_LEVEL=INFO,
REGION=us-east-1
}'
# Via SAM template ([Link])
Environment:
Variables:
TABLE_NAME: !Ref MyTable
BUCKET_NAME: !Ref MyBucket
LOG_LEVEL: INFO
Reading Env Vars in Python
python
import os
import logging
# Read at module level — evaluated once during Init phase
TABLE_NAME = [Link]['TABLE_NAME'] # raises KeyError if missing
BUCKET_NAME = [Link]('BUCKET_NAME', '') # safe default
LOG_LEVEL = [Link]('LOG_LEVEL', 'INFO')
REGION = [Link]('AWS_REGION', 'us-east-1') # auto-set by Lambda
logger = [Link]()
[Link](getattr(logging, LOG_LEVEL, [Link]))
def handler(event, context):
[Link]('Using table: %s', TABLE_NAME)
...
AWS Systems Manager Parameter Store
python
import boto3
import os
ssm = [Link]('ssm', region_name=[Link]['AWS_REGION'])
def get_parameter(name: str, decrypt: bool = True) -> str:
"""Fetch a parameter — cache the result to avoid repeated SSM calls."""
resp = ssm.get_parameter(Name=name, WithDecryption=decrypt)
return resp['Parameter']['Value']
# Cache at module level for warm reuse
_db_password = None
def get_db_password() -> str:
global _db_password
if _db_password is None:
_db_password = get_parameter('/myapp/prod/db_password')
return _db_password
AWS Lambda Complete Guide | Python Edition | 2025 Page 13
AWS 7 — ENVIRONMENT VARIABLES & CONFIGURATION AWS Lambda — Python Complete Guide
def handler(event, context):
password = get_db_password() # cached after first call
...
AWS Secrets Manager
python
import boto3
import json
import os
secrets_client = [Link]('secretsmanager')
def get_secret(secret_name: str) -> dict:
resp = secrets_client.get_secret_value(SecretId=secret_name)
return [Link](resp['SecretString'])
# Module-level cache
_secret = None
def get_credentials() -> dict:
global _secret
if _secret is None:
_secret = get_secret('myapp/production/db-credentials')
return _secret
def handler(event, context):
creds = get_credentials()
db_user = creds['username']
db_pass = creds['password']
...
WARNING
Never hard-code secrets in your source code or Lambda environment variables (they appear in the console). Use
Secrets Manager or Parameter Store (SecureString) and grant Lambda access via its execution role.
AWS Lambda Complete Guide | Python Edition | 2025 Page 14
AWS 8 — LAMBDA LAYERS AWS Lambda — Python Complete Guide
8 Lambda Layers
What are Layers?
A Lambda Layer is a .zip archive containing libraries, a custom runtime, or other dependencies. You can attach up to 5 layers to
a function. Layers are extracted to /opt in the execution environment, reducing deployment package size and enabling code
sharing across functions.
Creating a Python Dependency Layer
bash
# 1. Create the layer package structure
mkdir -p layer/python/lib/python3.12/site-packages
# 2. Install dependencies into the layer directory
pip install requests pandas boto3 \
--target layer/python/lib/python3.12/site-packages
# 3. Zip the layer
cd layer
zip -r ../[Link] python/
cd ..
# 4. Publish the layer
aws lambda publish-layer-version \
--layer-name my-python-deps \
--description "requests, pandas, boto3" \
--zip-file fileb://[Link] \
--compatible-runtimes python3.12 python3.11
# Output includes LayerVersionArn — use this in your function config
Attaching a Layer to a Function
bash
# Via CLI
aws lambda update-function-configuration \
--function-name MyFunction \
--layers \
arn:aws:lambda:us-east-1:123456789:layer:my-python-deps:3 \
arn:aws:lambda:us-east-1:580247275435:layer:LambdaInsightsExtension:38
# Via SAM template
Layers:
- !Ref MyDepsLayer
- arn:aws:lambda:us-east-1:580247275435:layer:LambdaInsightsExtension:38
Shared Utility Layer Example
python
# layer/python/utils/__init__.py (available as 'from utils import ...')
import logging
import json
import os
from datetime import datetime, timezone
from typing import Any
def get_logger(name: str = __name__) -> [Link]:
logger = [Link](name)
[Link]([Link]('LOG_LEVEL', 'INFO'))
return logger
def success(body: Any, status: int = 200) -> dict:
AWS Lambda Complete Guide | Python Edition | 2025 Page 15
AWS 8 — LAMBDA LAYERS AWS Lambda — Python Complete Guide
return {
'statusCode': status,
'headers': {'Content-Type': 'application/json'},
'body': [Link](body, default=str),
}
def error(message: str, status: int = 500) -> dict:
return {
'statusCode': status,
'headers': {'Content-Type': 'application/json'},
'body': [Link]({'error': message}),
}
def utcnow() -> str:
return [Link]([Link]).isoformat()
# In your Lambda function (layer is on /opt, which is in [Link]):
# from utils import get_logger, success, error
TIP
Keep layer packages small. Lambda has a 250 MB unzipped limit per function (code + all layers). Use pip install
--no-deps to avoid transitive bloat.
AWS Lambda Complete Guide | Python Edition | 2025 Page 16
AWS 9 — WORKING WITH S3 AWS Lambda — Python Complete Guide
9 Working with S3 in Python
Common S3 Operations with boto3
python
import boto3
import json
import os
from io import BytesIO
s3 = [Link]('s3') # initialised at module level — reused on warm invocations
# ■■ Read object ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def read_json_from_s3(bucket: str, key: str) -> dict:
resp = s3.get_object(Bucket=bucket, Key=key)
return [Link](resp['Body'].read().decode('utf-8'))
# ■■ Write object ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def write_json_to_s3(bucket: str, key: str, data: dict) -> None:
s3.put_object(
Bucket=bucket,
Key=key,
Body=[Link](data).encode('utf-8'),
ContentType='application/json',
)
# ■■ List objects ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def list_keys(bucket: str, prefix: str = '') -> list[str]:
paginator = s3.get_paginator('list_objects_v2')
keys = []
for page in [Link](Bucket=bucket, Prefix=prefix):
for obj in [Link]('Contents', []):
[Link](obj['Key'])
return keys
# ■■ Copy / Move ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def move_object(bucket: str, src_key: str, dst_key: str) -> None:
s3.copy_object(
CopySource={'Bucket': bucket, 'Key': src_key},
Bucket=bucket,
Key=dst_key,
)
s3.delete_object(Bucket=bucket, Key=src_key)
# ■■ Delete ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def delete_object(bucket: str, key: str) -> None:
s3.delete_object(Bucket=bucket, Key=key)
# ■■ Generate presigned URL ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def presigned_url(bucket: str, key: str, expiry_secs: int = 3600) -> str:
return s3.generate_presigned_url(
'get_object',
Params={'Bucket': bucket, 'Key': key},
ExpiresIn=expiry_secs,
)
S3-Triggered Lambda — Image Processing Example
python
import boto3
import os
from [Link] import unquote_plus
AWS Lambda Complete Guide | Python Edition | 2025 Page 17
AWS 9 — WORKING WITH S3 AWS Lambda — Python Complete Guide
s3 = [Link]('s3')
DEST_BUCKET = [Link]['DEST_BUCKET']
def handler(event, context):
"""Triggered by S3 ObjectCreated — process uploaded images."""
for record in event['Records']:
src_bucket = record['s3']['bucket']['name']
src_key = unquote_plus(record['s3']['object']['key'])
# Skip if not an image
if not src_key.lower().endswith(('.jpg', '.jpeg', '.png')):
continue
# Download to /tmp (ephemeral storage)
local_path = f'/tmp/{[Link](src_key)}'
s3.download_file(src_bucket, src_key, local_path)
# Process (resize, watermark, etc.)
output_path = process_image(local_path)
# Upload result
dest_key = f'processed/{[Link](output_path)}'
s3.upload_file(output_path, DEST_BUCKET, dest_key)
print(f'Processed {src_key} -> s3://{DEST_BUCKET}/{dest_key}')
def process_image(path: str) -> str:
# Placeholder — use Pillow layer for actual image processing
return path
AWS Lambda Complete Guide | Python Edition | 2025 Page 18
AWS 10 — WORKING WITH DYNAMODB AWS Lambda — Python Complete Guide
10 Working with DynamoDB in Python
DynamoDB CRUD with boto3
python
import boto3
import os
from [Link] import Key, Attr
from decimal import Decimal
dynamodb = [Link]('dynamodb') # higher-level resource API
table = [Link]([Link]['TABLE_NAME'])
# ■■ Create / Put Item ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def create_user(user_id: str, name: str, email: str) -> None:
table.put_item(Item={
'PK': f'USER#{user_id}',
'SK': 'PROFILE',
'name': name,
'email': email,
'created_at': '2025-01-01T00:00:00Z',
})
# ■■ Read / Get Item ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def get_user(user_id: str) -> dict | None:
resp = table.get_item(Key={'PK': f'USER#{user_id}', 'SK': 'PROFILE'})
return [Link]('Item') # None if not found
# ■■ Update Item ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def update_email(user_id: str, new_email: str) -> None:
table.update_item(
Key={'PK': f'USER#{user_id}', 'SK': 'PROFILE'},
UpdateExpression='SET email = :email, updated_at = :ts',
ExpressionAttributeValues={
':email': new_email,
':ts': '2025-06-01T12:00:00Z',
},
)
# ■■ Delete Item ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def delete_user(user_id: str) -> None:
table.delete_item(Key={'PK': f'USER#{user_id}', 'SK': 'PROFILE'})
# ■■ Query (by partition key, optionally sort key) ■■■■■■■■■■■■■■■■■■■■■
def get_user_orders(user_id: str) -> list:
resp = [Link](
KeyConditionExpression=Key('PK').eq(f'USER#{user_id}') &
Key('SK').begins_with('ORDER#'),
)
return [Link]('Items', [])
# ■■ Scan (full table — use sparingly) ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def find_active_users() -> list:
resp = [Link](FilterExpression=Attr('status').eq('active'))
return [Link]('Items', [])
# ■■ Batch Write ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def bulk_insert(users: list[dict]) -> None:
with table.batch_writer() as batch:
for user in users:
batch.put_item(Item=user)
DynamoDB Streams — Change Data Capture
python
import
AWS Lambdajson
Complete Guide | Python Edition | 2025 Page 19
AWS 10 — WORKING WITH DYNAMODB AWS Lambda — Python Complete Guide
def streams_handler(event, context):
"""Process DynamoDB stream records — NEW_AND_OLD_IMAGES."""
for record in event['Records']:
event_name = record['eventName'] # INSERT | MODIFY | REMOVE
new_image = [Link]('dynamodb', {}).get('NewImage', {})
old_image = [Link]('dynamodb', {}).get('OldImage', {})
# Deserialize DynamoDB AttributeValue format
new_item = deserialize(new_image)
old_item = deserialize(old_image)
if event_name == 'INSERT':
on_insert(new_item)
elif event_name == 'MODIFY':
on_update(old_item, new_item)
elif event_name == 'REMOVE':
on_delete(old_item)
def deserialize(dynamo_item: dict) -> dict:
from [Link] import TypeDeserializer
deserializer = TypeDeserializer()
return {k: [Link](v) for k, v in dynamo_item.items()}
def on_insert(item): print('Inserted:', item)
def on_update(old, new): print('Updated:', old, '->', new)
def on_delete(item): print('Deleted:', item)
AWS Lambda Complete Guide | Python Edition | 2025 Page 20
AWS 11 — SQS & SNS AWS Lambda — Python Complete Guide
11 Working with SQS & SNS
Processing SQS Messages
python
import json
import boto3
import os
from typing import Any
sqs = [Link]('sqs')
QUEUE_URL = [Link]['QUEUE_URL']
# ■■ Lambda triggered by SQS ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def sqs_handler(event, context) -> dict:
"""Handle SQS batch — return partial failure response."""
batch_failures = []
for record in event['Records']:
message_id = record['messageId']
receipt = record['receiptHandle']
body = [Link](record['body'])
try:
process_order(body)
# Successful messages are auto-deleted by Lambda's SQS trigger
except Exception as exc:
print(f'ERROR processing {message_id}: {exc}')
# Return to queue for retry (with partial batch response)
batch_failures.append({'itemIdentifier': message_id})
# Partial batch response: only failed messages become visible again
return {'batchItemFailures': batch_failures}
def process_order(order: dict) -> None:
print(f'Processing order {order["id"]}')
# Your business logic here
# ■■ Send a message to SQS ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def send_message(body: Any, delay_seconds: int = 0) -> str:
resp = sqs.send_message(
QueueUrl=QUEUE_URL,
MessageBody=[Link](body),
DelaySeconds=delay_seconds,
)
return resp['MessageId']
# ■■ Send batch of messages ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def send_batch(messages: list[Any]) -> None:
entries = [
{'Id': str(i), 'MessageBody': [Link](msg)}
for i, msg in enumerate(messages)
]
# SendMessageBatch accepts up to 10 messages
for i in range(0, len(entries), 10):
sqs.send_message_batch(QueueUrl=QUEUE_URL, Entries=entries[i:i+10])
Publishing to SNS
python
import boto3
import json
AWS Lambda Complete Guide | Python Edition | 2025 Page 21
AWS 11 — SQS & SNS AWS Lambda — Python Complete Guide
import os
sns = [Link]('sns')
TOPIC_ARN = [Link]['SNS_TOPIC_ARN']
def publish_event(subject: str, message: dict) -> str:
"""Publish a JSON message to an SNS topic."""
resp = [Link](
TopicArn=TOPIC_ARN,
Subject=subject,
Message=[Link](message),
MessageAttributes={
'event_type': {
'DataType': 'String',
'StringValue': [Link]('type', 'unknown'),
}
},
)
return resp['MessageId']
# ■■ SNS -> Lambda trigger ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def sns_handler(event, context):
"""Lambda triggered by SNS subscription."""
for record in event['Records']:
sns_msg = record['Sns']
subject = sns_msg.get('Subject', '')
body = [Link](sns_msg['Message'])
attrs = sns_msg.get('MessageAttributes', {})
print(f'SNS message: subject={subject}, body={body}')
handle_notification(subject, body)
def handle_notification(subject: str, body: dict) -> None:
if subject == 'ORDER_PLACED':
send_confirmation_email(body['user_email'], body['order_id'])
AWS Lambda Complete Guide | Python Edition | 2025 Page 22
AWS 12 — API GATEWAY + LAMBDA AWS Lambda — Python Complete Guide
12 API Gateway + Lambda — REST APIs
HTTP API (v2) Handler Pattern
python
import json
import boto3
import os
from typing import Any
dynamodb = [Link]('dynamodb')
table = [Link]([Link]['TABLE_NAME'])
# ■■ Response helpers ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def resp(status: int, body: Any) -> dict:
return {
'statusCode': status,
'headers': {
'Content-Type': 'application/json',
'Access-Control-Allow-Origin': '*', # CORS
},
'body': [Link](body, default=str),
}
# ■■ Router ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
ROUTES = {}
def route(method: str, path: str):
"""Decorator to register route handlers."""
def decorator(fn):
ROUTES[([Link](), path)] = fn
return fn
return decorator
# ■■ Route handlers ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
@route('GET', '/users')
def list_users(event):
items = [Link]().get('Items', [])
return resp(200, items)
@route('GET', '/users/{id}')
def get_user(event):
user_id = event['pathParameters']['id']
item = table.get_item(Key={'PK': f'USER#{user_id}'}).get('Item')
if not item:
return resp(404, {'error': 'User not found'})
return resp(200, item)
@route('POST', '/users')
def create_user(event):
body = [Link]([Link]('body') or '{}')
if not [Link]('email'):
return resp(400, {'error': 'email is required'})
table.put_item(Item={'PK': f'USER#{body["id"]}', **body})
return resp(201, body)
@route('DELETE', '/users/{id}')
def delete_user(event):
user_id = event['pathParameters']['id']
table.delete_item(Key={'PK': f'USER#{user_id}'})
return resp(204, {})
# ■■ Main handler ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
AWS Lambda Complete Guide | Python Edition | 2025 Page 23
AWS 12 — API GATEWAY + LAMBDA AWS Lambda — Python Complete Guide
def handler(event, context):
method = event['requestContext']['http']['method']
path = event['routeKey'].split(' ', 1)[-1] # e.g. '/users/{id}'
handler_fn = [Link]((method, path))
if not handler_fn:
return resp(404, {'error': f'Route {method} {path} not found'})
try:
return handler_fn(event)
except Exception as exc:
print(f'Unhandled error: {exc}')
return resp(500, {'error': 'Internal server error'})
SAM Template for HTTP API
yaml
# [Link]
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Globals:
Function:
Runtime: python3.12
MemorySize: 256
Timeout: 30
Environment:
Variables:
TABLE_NAME: !Ref UsersTable
Resources:
ApiFunction:
Type: AWS::Serverless::Function
Properties:
Handler: lambda_function.handler
Events:
ApiEvent:
Type: HttpApi
Properties:
Path: /{proxy+}
Method: ANY
UsersTable:
Type: AWS::DynamoDB::Table
Properties:
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- { AttributeName: PK, AttributeType: S }
KeySchema:
- { AttributeName: PK, KeyType: HASH }
AWS Lambda Complete Guide | Python Edition | 2025 Page 24
AWS 13 — LAMBDA INSIDE A VPC AWS Lambda — Python Complete Guide
13 Lambda inside a VPC
VPC Architecture
Lambda inside a VPC
Internet
Your VPC
NAT Gateway
Lambda RDS / Redis
Private Subnet Private Subnet
When to Use VPC
• Your Lambda needs to connect to an RDS, Aurora, or ElastiCache cluster in private subnets.
• You need to call internal services or APIs that are not publicly accessible.
• Compliance requires workloads to be isolated within a private network.
• You use a self-hosted database or message broker on EC2.
WARNING
Lambdas inside a VPC lose internet access by default. Add a NAT Gateway in a public subnet (not NAT instance — too
slow) for outbound internet / AWS API calls.
VPC Configuration
bash
# Via CLI
aws lambda update-function-configuration \
--function-name MyFunction \
--vpc-config SubnetIds=subnet-aaa,subnet-bbb,SecurityGroupIds=sg-xyz
# In SAM template
VpcConfig:
SubnetIds:
- !Ref PrivateSubnetA
- !Ref PrivateSubnetB
SecurityGroupIds:
- !Ref LambdaSecurityGroup
Connecting to RDS (PostgreSQL) from Lambda
python
import boto3
import os
import json
import psycopg2 # install via layer or in deployment package
secrets_client = [Link]('secretsmanager')
_conn = None # module-level connection — reused on warm invocations
def get_connection():
global _conn
if _conn is None or _conn.closed:
secret = [Link](
secrets_client.get_secret_value(
SecretId=[Link]['DB_SECRET_ARN']
AWS Lambda Complete Guide | Python Edition | 2025 Page 25
AWS 13 — LAMBDA INSIDE A VPC AWS Lambda — Python Complete Guide
)['SecretString']
)
_conn = [Link](
host=[Link]['DB_HOST'],
port=int([Link]('DB_PORT', 5432)),
database=[Link]['DB_NAME'],
user=secret['username'],
password=secret['password'],
connect_timeout=5,
# Use SSL — required for RDS
sslmode='require',
)
return _conn
def handler(event, context):
conn = get_connection()
with [Link]() as cur:
[Link]('SELECT id, name FROM users WHERE active = %s', (True,))
rows = [Link]()
return {'statusCode': 200, 'body': [Link](rows)}
RDS Proxy — Recommended for Lambda + RDS
Lambda can open thousands of concurrent connections to RDS, exhausting the connection pool. RDS Proxy sits between
Lambda and RDS, pooling and reusing connections. It also handles failover transparently.
python
# With RDS Proxy, point your DB_HOST to the proxy endpoint instead of RDS
# e.g. DB_HOST = [Link]
# RDS Proxy uses IAM authentication (recommended — no password needed)
import boto3
import os
rds_client = [Link]('rds')
def get_rds_iam_token() -> str:
return rds_client.generate_db_auth_token(
DBHostname=[Link]['DB_HOST'],
Port=5432,
DBUsername=[Link]['DB_USER'],
Region=[Link]['AWS_REGION'],
)
AWS Lambda Complete Guide | Python Edition | 2025 Page 26
AWS 14 — ERROR HANDLING, RETRIES & DLQ AWS Lambda — Python Complete Guide
14 Error Handling, Retries & DLQ
Lambda Error Handling Fundamentals
python
import json
import logging
logger = [Link]()
class ValidationError(Exception):
"""Custom exception — signals bad input, should NOT be retried."""
class TransientError(Exception):
"""Custom exception — signals transient failure, retry is OK."""
def handler(event, context):
try:
result = process(event)
return {'statusCode': 200, 'body': [Link](result)}
except ValidationError as exc:
# 4xx — client error, no retry needed
[Link]('Validation failed: %s', exc)
return {'statusCode': 400, 'body': [Link]({'error': str(exc)})}
except TransientError as exc:
# 5xx — transient; Lambda may retry (async) or SQS re-enqueues
[Link]('Transient error: %s', exc)
raise # re-raise so Lambda marks invocation as failed
except Exception as exc:
# Unexpected error — always log and re-raise
[Link]('Unexpected error processing event: %s', event)
raise
def process(event: dict) -> dict:
if not [Link]('id'):
raise ValidationError('id is required')
# ... business logic
return {'processed': True}
Retry Behaviour by Invocation Type
Invocation Type Retries Configurable? DLQ Support
Synchronous (API GW) 0 (error returned to caller) No No
Asynchronous (S3, SNS) 2 automatic retries (3 total attempts) Yes (0-2) Yes — SQS or SNS
SQS Up to maxReceiveCount (redrive policy) Yes Yes — SQS DLQ
Kinesis / DynamoDB Streams Until success or data expiry (24h–7d) Yes — bisect-on-errorYes — S3 / SQS / SNS
Configuring a Dead Letter Queue
python
# Attach SQS DLQ to function (async invocations)
aws lambda update-function-configuration \
--function-name MyFunction \
--dead-letter-config TargetArn=arn:aws:sqs:us-east-1:123456789:MyDLQ
# In SAM template
DeadLetterQueue:
Type: SQS
AWS Lambda Complete Guide | Python Edition | 2025 Page 27
AWS 14 — ERROR HANDLING, RETRIES & DLQ AWS Lambda — Python Complete Guide
TargetArn: !GetAtt [Link]
# Process DLQ messages (alert, debug, reprocess)
def dlq_handler(event, context):
for record in event['Records']:
original_event = [Link](record['body'])
error_info = {
'error_message': record['attributes'].get('ErrorMessage'),
'error_code': record['attributes'].get('ErrorCode'),
'approximate_first_receive': record['attributes'].get('ApproximateFirstReceiveTimestamp'),
}
[Link]('DLQ message: %s | Error: %s', original_event, error_info)
# Alert ops team, store to S3 for analysis, etc.
Lambda Destinations
Lambda Destinations & Error Handling
On Success
SQS / SNS / Lambda / EB
DLQ / Retry Lambda
max 2 retries (async)
On Failure
DLQ / SQS / SNS / EB
AWS Lambda Complete Guide | Python Edition | 2025 Page 28
AWS 15 — LAMBDA DESTINATIONS AWS Lambda — Python Complete Guide
15 Lambda Destinations (Async Routing)
Configuring Destinations
bash
# Set destinations via CLI
aws lambda put-function-event-invoke-config \
--function-name MyFunction \
--maximum-retry-attempts 1 \
--destination-config '{
"OnSuccess": {
"Destination": "arn:aws:sqs:us-east-1:123456789:SuccessQueue"
},
"OnFailure": {
"Destination": "arn:aws:sns:us-east-1:123456789:AlertTopic"
}
}'
# In SAM template
EventInvokeConfig:
MaximumRetryAttempts: 1
DestinationConfig:
OnSuccess:
Type: SQS
Destination: !GetAtt [Link]
OnFailure:
Type: SNS
Destination: !Ref AlertTopic
Destination Payload Structure
json
# Lambda sends this JSON to your destination (SQS/SNS/Lambda/EventBridge)
{
"version": "1.0",
"timestamp": "2025-01-01T12:00:00.000Z",
"requestContext": {
"requestId": "abc-123",
"functionArn": "arn:aws:lambda:us-east-1:123:function:MyFunction",
"condition": "Success", # or "RetriesExhausted"
"approximateInvokeCount": 1
},
"requestPayload": { ... }, # original event your function received
"responseContext": {
"statusCode": 200,
"executedVersion": "$LATEST"
},
"responsePayload": { ... } # what your function returned (on success)
}
# On failure, responsePayload contains the error:
{
"responsePayload": {
"errorMessage": "...",
"errorType": "RuntimeError",
"stackTrace": [...]
}
}
Step Functions as Destination (Orchestration)
For complex multi-step workflows, use AWS Step Functions with Lambda as task states. This gives you visual workflows, built-in
retry logic, parallel execution, and state management without managing queues manually.
json
# Lambda task in a Step Functions state machine
AWS Lambda Complete Guide | Python Edition | 2025 Page 29
AWS 15 — LAMBDA DESTINATIONS AWS Lambda — Python Complete Guide
{
"Comment": "Order processing workflow",
"StartAt": "ValidateOrder",
"States": {
"ValidateOrder": {
"Type": "Task",
"Resource": "arn:aws:lambda:us-east-1:123:function:ValidateOrderFn",
"Next": "ChargePayment",
"Retry": [{"ErrorEquals": ["[Link]"], "MaxAttempts": 2}],
"Catch": [{"ErrorEquals": ["[Link]"], "Next": "HandleError"}]
},
"ChargePayment": {
"Type": "Task",
"Resource": "arn:aws:lambda:us-east-1:123:function:ChargePaymentFn",
"Next": "SendConfirmation"
},
"SendConfirmation": {
"Type": "Task",
"Resource": "arn:aws:lambda:us-east-1:123:function:SendEmailFn",
"End": true
},
"HandleError": { "Type": "Fail" }
}
}
AWS Lambda Complete Guide | Python Edition | 2025 Page 30
AWS 16 — OBSERVABILITY AWS Lambda — Python Complete Guide
16 Observability — Logs, Metrics & Tracing
Structured Logging with Python
python
import json
import logging
import os
import time
from typing import Any
logger = [Link]()
[Link]([Link]('LOG_LEVEL', 'INFO'))
class StructuredLogger:
"""Emit JSON log lines for easy CloudWatch Insights queries."""
def __init__(self, function_name: str):
self.function_name = function_name
def _log(self, level: str, message: str, **extra: Any) -> None:
record = {
'level': level,
'message': message,
'function': self.function_name,
'timestamp': [Link]('%Y-%m-%dT%H:%M:%SZ', [Link]()),
**extra
}
print([Link](record)) # Lambda captures stdout to CloudWatch
def info(self, msg, **kw): self._log('INFO', msg, **kw)
def warn(self, msg, **kw): self._log('WARN', msg, **kw)
def error(self, msg, **kw): self._log('ERROR', msg, **kw)
log = StructuredLogger([Link]('AWS_LAMBDA_FUNCTION_NAME', 'local'))
def handler(event, context):
[Link]('Invocation started', request_id=context.aws_request_id,
event_keys=list([Link]()))
try:
result = do_work(event)
[Link]('Invocation succeeded', result_size=len(str(result)))
return result
except Exception as exc:
[Link]('Invocation failed', error=str(exc), error_type=type(exc).__name__)
raise
Custom CloudWatch Metrics
python
import boto3
import os
cloudwatch = [Link]('cloudwatch')
NAMESPACE = [Link]('METRICS_NAMESPACE', 'MyApp')
def put_metric(name: str, value: float, unit: str = 'Count',
dimensions: dict | None = None) -> None:
"""Publish a custom metric to CloudWatch."""
dim_list = [{'Name': k, 'Value': v} for k, v in (dimensions or {}).items()]
cloudwatch.put_metric_data(
Namespace=NAMESPACE,
MetricData=[{
AWS Lambda Complete Guide | Python Edition | 2025 Page 31
AWS 16 — OBSERVABILITY AWS Lambda — Python Complete Guide
'MetricName': name,
'Value': value,
'Unit': unit,
'Dimensions': dim_list,
}],
)
def handler(event, context):
start = [Link]()
orders_processed = 0
for record in [Link]('Records', []):
process_record(record)
orders_processed += 1
duration_ms = ([Link]() - start) * 1000
put_metric('OrdersProcessed', orders_processed, 'Count',
{'Stage': [Link]('STAGE', 'prod')})
put_metric('ProcessingLatency', duration_ms, 'Milliseconds')
AWS X-Ray Distributed Tracing
python
# Enable X-Ray active tracing in SAM template:
# Tracing: Active
# Install: pip install aws-xray-sdk
from aws_xray_sdk.core import xray_recorder, patch_all
# Patch supported libraries (boto3, requests, psycopg2, etc.)
patch_all()
@xray_recorder.capture('process_payment')
def process_payment(order_id: str, amount: float) -> dict:
# This function gets its own X-Ray subsegment
xray_recorder.current_subsegment().put_annotation('order_id', order_id)
xray_recorder.current_subsegment().put_metadata('amount', amount)
# ... payment logic
return {'status': 'charged'}
def handler(event, context):
with xray_recorder.in_subsegment('validate_input') as sub:
sub.put_annotation('event_source', [Link]('source', 'unknown'))
# validation logic
result = process_payment(event['order_id'], event['amount'])
return {'statusCode': 200, 'body': [Link](result)}
Useful CloudWatch Insights Queries
bash
# Find all errors in last 1 hour
fields @timestamp, @message
| filter @message like /ERROR/
| sort @timestamp desc
| limit 50
# Calculate p99 duration for cold starts
filter @type = "REPORT"
| stats pct(@duration, 99) as p99, avg(@duration) as avg by bin(5m)
# Find timeouts
AWS Lambda Complete Guide | Python Edition | 2025 Page 32
AWS 16 — OBSERVABILITY AWS Lambda — Python Complete Guide
filter @message like /Task timed out/
| stats count(*) by bin(1h)
# Find memory usage (identify over/under-allocated functions)
filter @type = "REPORT"
| stats max(@maxMemoryUsed / 1000 / 1000) as maxMemMB,
avg(@maxMemoryUsed / 1000 / 1000) as avgMemMB
by @logStream
AWS Lambda Complete Guide | Python Edition | 2025 Page 33
AWS 17 — PERFORMANCE TUNING AWS Lambda — Python Complete Guide
17 Performance — Memory, Concurrency & SnapStart
Memory & CPU Allocation
Lambda allocates CPU proportionally to memory. More memory = more vCPU. Increasing memory often reduces execution time
enough to lower total cost. Always benchmark!
Memory Approx vCPU Use Case
128 MB 0.08 vCPU Lightweight: simple transforms, notifications
256 MB 0.16 vCPU Light APIs, basic data processing
512 MB 0.33 vCPU Typical APIs, moderate computation
1024 MB 0.67 vCPU ML inference, image processing
1769 MB 1 vCPU One full vCPU — ideal for CPU-bound tasks
3008 MB 2 vCPU Heavy parallel processing
10240 MB 6 vCPU Maximum — large in-memory datasets
Concurrency Models
Concurrency Type Description Cost
Unreserved Default pool shared by all functions in region — no guarante Standard Lambda pricing
Reserved Guaranteed maximum for one function — throttles above limit No extra charge
Provisioned Pre-initialised environments — eliminates cold starts entire +approx 15% of on-demand price
bash
# Reserve concurrency (cap max simultaneous executions)
aws lambda put-function-concurrency \
--function-name MyFunction \
--reserved-concurrent-executions 100
# Provisioned concurrency (pre-warm N environments)
aws lambda put-provisioned-concurrency-config \
--function-name MyFunction \
--qualifier prod \ # must use an alias or version
--provisioned-concurrent-executions 10
# Auto-scaling provisioned concurrency (scale with traffic)
aws application-autoscaling register-scalable-target \
--service-namespace lambda \
--resource-id function:MyFunction:prod \
--scalable-dimension lambda:function:ProvisionedConcurrency \
--min-capacity 5 --max-capacity 50
Performance Best Practices
• Right-size memory: run AWS Lambda Power Tuning (open-source Step Functions tool) to find the cost-optimal memory
setting.
• Reuse connections: initialise boto3 clients, DB connections, and HTTP sessions at module level.
• Use connection pooling with RDS Proxy to avoid connection exhaustion.
• Reduce package size: use slim base layers, avoid unused dependencies, enable tree-shaking.
• Use /tmp for caching: up to 10 GB, persists across warm invocations for the same environment.
• Batch SQS messages: larger batch sizes (up to 10,000) reduce per-record overhead.
• Prefer async invocation for non-latency-sensitive work to decouple caller and Lambda.
• Set timeout conservatively: just above expected maximum duration to avoid zombie executions.
AWS Lambda Complete Guide | Python Edition | 2025 Page 34
AWS 18 — EXTENSIONS & SNAPSTART AWS Lambda — Python Complete Guide
18 Lambda Extensions & SnapStart
Lambda Extensions
Lambda Extensions are companion processes that run alongside your function code in the same execution environment. They
can intercept lifecycle events (Init, Invoke, Shutdown) for tasks like telemetry collection, secret caching, or custom monitoring.
Extension Type Description Examples
Internal extension Runs in same process as function (e.g. via LAMBDA_HANDLER
Custom logging
en wrappers
External extension Separate process in /opt/extensions/ — runs in parallel to
Datadog,
f New Relic, AWS CloudWatch Lambda Insights, Hash
Lambda Insights Extension
yaml
# Enable Lambda Insights (enhanced metrics) via SAM
Layers:
- !Sub arn:aws:lambda:${AWS::Region}:580247275435:layer:LambdaInsightsExtension:38
# Requires CloudWatch Lambda Insights policy on execution role:
# arn:aws:iam::aws:policy/CloudWatchLambdaInsightsExecutionRolePolicy
# Metrics it adds:
# - init_duration (cold start duration)
# - cpu_total_time
# - memory_utilization
# - rx_bytes / tx_bytes (network)
# - fd_use / fd_max (file descriptors)
SnapStart (Java; Python coming)
SnapStart takes a snapshot of a fully initialised execution environment and restores it on demand, eliminating cold start latency.
Currently available for Java (Corretto 11+) with Python support expected. When available for Python, enable it like this:
python
# SAM template — enable SnapStart (Java; apply to Python when available)
MyFunction:
Type: AWS::Serverless::Function
Properties:
SnapStart:
ApplyOn: PublishedVersions
AutoPublishAlias: live
# SnapStart best practices for your code:
# 1. Avoid storing timestamps/random seeds at init time (restored snapshot has stale values)
# 2. Re-seed random number generators in the handler, not at module level
# 3. Re-establish network connections in the handler (connections may be stale after restore)
# 4. Implement the AfterRestore hook to refresh state after snapshot restore
# Correct pattern for SnapStart-compatible code
import random
import boto3
# BAD: initialised at module level — snapshot captures stale timestamp
STARTED_AT = [Link]() # stale after restore!
# GOOD: lazy init inside handler
_rng = None
def get_rng():
global _rng
if _rng is None:
_rng = [Link]() # fresh on first call after restore
return _rng
AWS Lambda Complete Guide | Python Edition | 2025 Page 35
AWS 19 — DEPLOYING WITH SAM AWS Lambda — Python Complete Guide
19 AWS SAM — Serverless Application Model
SAM Workflow
SAM / CDK Deployment Workflow
Write sam build sam local sam deploy CloudFormation Live
Code / cdk synth invoke / cdk deploy Stack Lambda
Install & Bootstrap
bash
# Install SAM CLI
brew install aws-sam-cli # macOS
# Linux: see [Link]
# Initialise new project from template
sam init --runtime python3.12
# Project layout created by sam init
my-app/
hello_world/
__init__.py
[Link] # handler
[Link]
tests/unit/test_handler.py
[Link] # SAM template
[Link] # saved deploy config (auto-created on first deploy)
Complete SAM Template
yaml
# [Link]
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: My serverless application
Globals:
Function:
Runtime: python3.12
Handler: [Link]
MemorySize: 512
Timeout: 30
Tracing: Active # X-Ray
Layers: [!Ref UtilsLayer]
Environment:
Variables:
TABLE_NAME: !Ref AppTable
LOG_LEVEL: INFO
Parameters:
Stage:
Type: String
Default: dev
AllowedValues: [dev, staging, prod]
Resources:
# ■■ API + Function ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
ApiFunction:
AWS Lambda Complete Guide | Python Edition | 2025 Page 36
AWS 19 — DEPLOYING WITH SAM AWS Lambda — Python Complete Guide
Type: AWS::Serverless::Function
Properties:
CodeUri: src/api/
Events:
Api:
Type: HttpApi
Properties:
Path: /{proxy+}
Method: ANY
Policies:
- DynamoDBCrudPolicy: { TableName: !Ref AppTable }
- S3ReadPolicy: { BucketName: !Ref AssetsBucket }
# ■■ Worker Function ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
WorkerFunction:
Type: AWS::Serverless::Function
Properties:
CodeUri: src/worker/
MemorySize: 1024
ReservedConcurrentExecutions: 50
Events:
Queue:
Type: SQS
Properties:
Queue: !GetAtt [Link]
BatchSize: 10
FunctionResponseTypes: [ReportBatchItemFailures]
DeadLetterQueue:
Type: SQS
TargetArn: !GetAtt [Link]
# ■■ Shared Layer ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
UtilsLayer:
Type: AWS::Serverless::LayerVersion
Properties:
ContentUri: layers/utils/
CompatibleRuntimes: [python3.12]
Metadata:
BuildMethod: python3.12
# ■■ DynamoDB Table ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
AppTable:
Type: AWS::DynamoDB::Table
DeletionPolicy: Retain
Properties:
BillingMode: PAY_PER_REQUEST
AttributeDefinitions:
- { AttributeName: PK, AttributeType: S }
- { AttributeName: SK, AttributeType: S }
KeySchema:
- { AttributeName: PK, KeyType: HASH }
- { AttributeName: SK, KeyType: RANGE }
PointInTimeRecoverySpecification:
PointInTimeRecoveryEnabled: true
WorkQueue:
Type: AWS::SQS::Queue
Properties:
VisibilityTimeout: 120
RedrivePolicy:
deadLetterTargetArn: !GetAtt [Link]
maxReceiveCount: 3
AWS Lambda Complete Guide | Python Edition | 2025 Page 37
AWS 19 — DEPLOYING WITH SAM AWS Lambda — Python Complete Guide
DeadLetterQueue:
Type: AWS::SQS::Queue
Properties:
MessageRetentionPeriod: 1209600 # 14 days
AssetsBucket:
Type: AWS::S3::Bucket
Outputs:
ApiUrl:
Value: !Sub [Link]
TableName:
Value: !Ref AppTable
SAM Build, Local Test & Deploy
bash
# Build (installs dependencies, packages code)
sam build
# Local invoke (uses Docker to simulate Lambda runtime)
sam local invoke ApiFunction --event events/api_event.json
# Local HTTP API (hot-reloading dev server)
sam local start-api --warm-containers EAGER --port 3000
# Run unit tests
python -m pytest tests/ -v
# Deploy (first time — creates guided config)
sam deploy --guided
# Deploy (subsequent — uses [Link])
sam deploy
# Validate template
sam validate
# View logs from deployed function
sam logs -n ApiFunction --stack-name my-app --tail
# Delete stack
sam delete --stack-name my-app
AWS Lambda Complete Guide | Python Edition | 2025 Page 38
AWS 20 — DEPLOYING WITH AWS CDK (PYTHON) AWS Lambda — Python Complete Guide
20 AWS CDK — Python Infrastructure as Code
CDK Setup
bash
# Install CDK CLI
npm install -g aws-cdk
# Create new Python CDK project
mkdir my-lambda-cdk && cd my-lambda-cdk
cdk init app --language python
# Install Python dependencies
python -m venv .venv
source .venv/bin/activate
pip install aws-cdk-lib constructs
# Bootstrap (one-time per account/region)
cdk bootstrap a[Link]
Full CDK Stack — Python
python
# [Link]
import aws_cdk as cdk
from my_stack import MyLambdaStack
app = [Link]()
MyLambdaStack(app, 'MyLambdaStack',
env=[Link](account='123456789', region='us-east-1'))
[Link]()
# my_stack.py
from aws_cdk import (
Stack, Duration, RemovalPolicy,
aws_lambda as _lambda,
aws_lambda_event_sources as event_sources,
aws_dynamodb as dynamodb,
aws_sqs as sqs,
aws_s3 as s3,
aws_iam as iam,
aws_apigatewayv2 as apigwv2,
aws_apigatewayv2_integrations as integrations,
aws_logs as logs,
)
from constructs import Construct
class MyLambdaStack(Stack):
def __init__(self, scope: Construct, id: str, **kwargs):
super().__init__(scope, id, **kwargs)
# ■■ DynamoDB Table ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
table = [Link](self, 'AppTable',
partition_key=[Link](
name='PK', type=[Link]),
sort_key=[Link](
name='SK', type=[Link]),
billing_mode=[Link].PAY_PER_REQUEST,
removal_policy=[Link],
point_in_time_recovery=True,
)
AWS Lambda Complete Guide | Python Edition | 2025 Page 39
AWS 20 — DEPLOYING WITH AWS CDK (PYTHON) AWS Lambda — Python Complete Guide
# ■■ SQS Queue ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
dlq = [Link](self, 'DLQ',
retention_period=[Link](14))
queue = [Link](self, 'WorkQueue',
visibility_timeout=[Link](120),
dead_letter_queue=[Link](
queue=dlq, max_receive_count=3))
# ■■ Shared Layer ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
utils_layer = _lambda.LayerVersion(self, 'UtilsLayer',
code=_lambda.Code.from_asset('layers/utils'),
compatible_runtimes=[_lambda.Runtime.PYTHON_3_12],
description='Shared utilities')
# ■■ Common Lambda config ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
common = dict(
runtime=_lambda.Runtime.PYTHON_3_12,
memory_size=512,
timeout=[Link](30),
tracing=_lambda.[Link],
layers=[utils_layer],
log_retention=[Link].ONE_MONTH,
environment={'TABLE_NAME': table.table_name, 'LOG_LEVEL': 'INFO'},
)
# ■■ API Function ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
api_fn = _lambda.Function(self, 'ApiFunction',
code=_lambda.Code.from_asset('src/api'),
handler='[Link]',
**common)
table.grant_read_write_data(api_fn)
http_api = [Link](self, 'HttpApi',
default_integration=[Link](
'ApiIntegration', api_fn))
# ■■ Worker Function ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
worker_fn = _lambda.Function(self, 'WorkerFunction',
code=_lambda.Code.from_asset('src/worker'),
handler='[Link]',
memory_size=1024,
reserved_concurrent_executions=50,
**{k: v for k, v in [Link]()
if k not in ('memory_size',)})
worker_fn.add_event_source(
event_sources.SqsEventSource(queue,
batch_size=10,
report_batch_item_failures=True))
table.grant_read_write_data(worker_fn)
CDK Commands
bash
cdk synth # synthesise CloudFormation template
cdk diff # show what will change
cdk deploy # deploy stack
cdk deploy --hotswap # fast deploy for Lambda code changes (dev only)
cdk watch # auto-deploy on file changes (dev)
cdk destroy # tear down stack
AWS Lambda Complete Guide | Python Edition | 2025 Page 40
AWS 21 — CI/CD FOR LAMBDA AWS Lambda — Python Complete Guide
21 CI/CD Pipeline — GitHub Actions
Complete GitHub Actions Workflow
yaml
# .github/workflows/[Link]
name: Lambda CI/CD
on:
push:
branches: [main]
pull_request:
branches: [main]
permissions:
id-token: write # OIDC — no long-lived AWS keys needed
contents: read
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with: { python-version: '3.12' }
- name: Install dependencies
run: pip install -r [Link]
- name: Lint
run: |
ruff check src/ tests/
mypy src/
- name: Run tests with coverage
run: pytest tests/ --cov=src --cov-report=xml -v
- name: Upload coverage
uses: codecov/codecov-action@v4
deploy:
needs: test
runs-on: ubuntu-latest
if: [Link] == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with: { python-version: '3.12' }
- name: Set up SAM CLI
uses: aws-actions/setup-sam@v2
- name: Configure AWS credentials (OIDC — no long-lived keys!)
uses: aws-actions/configure-aws-credentials@v4
with:
role-to-assume: arn:aws:iam::123456789:role/GitHubActionsDeployRole
aws-region: us-east-1
AWS Lambda Complete Guide | Python Edition | 2025 Page 41
AWS 21 — CI/CD FOR LAMBDA AWS Lambda — Python Complete Guide
- name: SAM Build
run: sam build --use-container
- name: Run integration tests
run: pytest tests/integration/ -v
env:
AWS_DEFAULT_REGION: us-east-1
- name: SAM Deploy
run: |
sam deploy \
--no-confirm-changeset \
--no-fail-on-empty-changeset \
--stack-name my-lambda-app-prod \
--parameter-overrides Stage=prod \
--tags Environment=prod Team=backend
OIDC Trust Policy (GitHub Actions -> AWS, no long-lived keys)
json
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {
"Federated": "arn:aws:iam::123456789:oidc-provider/[Link]"
},
"Action": "sts:AssumeRoleWithWebIdentity",
"Condition": {
"StringEquals": {
"[Link]:aud": "[Link]"
},
"StringLike": {
"[Link]:sub":
"repo:my-org/my-repo:ref:refs/heads/main"
}
}
}]
}
AWS Lambda Complete Guide | Python Edition | 2025 Page 42
AWS 22 — SECURITY BEST PRACTICES AWS Lambda — Python Complete Guide
22 Security Best Practices
Code-Level Security
python
import os
import json
import re
# ■■ Input validation ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
def validate_event(event: dict) -> dict:
"""Always validate and sanitise event input."""
body = [Link]([Link]('body') or '{}')
user_id = [Link]('user_id', '')
if not [Link](r'^[a-zA-Z0-9_-]{1,64}$', user_id):
raise ValueError(f'Invalid user_id: {user_id!r}')
amount = float([Link]('amount', 0))
if amount <= 0 or amount > 100_000:
raise ValueError(f'Invalid amount: {amount}')
return {'user_id': user_id, 'amount': amount}
# ■■ Parameterised queries (avoid SQL injection) ■■■■■■■■■■■■■■■■■■■■■■■
# BAD:
# [Link](f"SELECT * FROM users WHERE id = '{user_id}'")
# GOOD:
# [Link]("SELECT * FROM users WHERE id = %s", (user_id,))
# ■■ Never log sensitive data ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
# BAD:
# [Link]('Processing payment for card: %s', card_number)
# GOOD:
# [Link]('Processing payment for card ending: %s', card_number[-4:])
# ■■ Sanitise before returning to client ■■■■■■■■■■■■■■■■■■■■■■■■■■■■■■
INTERNAL_FIELDS = {'password_hash', 'internal_notes', 'admin_flag'}
def sanitise(item: dict) -> dict:
return {k: v for k, v in [Link]() if k not in INTERNAL_FIELDS}
Infrastructure Security Checklist
• Apply least privilege IAM — scope every permission to exact resources needed.
• Use OIDC for CI/CD — eliminate long-lived AWS access keys entirely.
• Enable VPC for functions accessing private resources (RDS, ElastiCache).
• Encrypt environment variables with a KMS Customer Managed Key (CMK).
• Use Secrets Manager or SSM SecureString — never plain-text secrets in env vars.
• Enable AWS CloudTrail to audit all Lambda API calls.
• Turn on GuardDuty Lambda Protection to detect anomalous invocations.
• Use Lambda function URLs with IAM auth instead of open HTTP endpoints.
• Set reserved concurrency to prevent a runaway function from exhausting account quota.
• Enable AWS Config rules to detect Lambda functions with overly permissive roles.
• Pin dependency versions in [Link] and use pip-audit for vulnerability scanning.
• Use code signing (AWS Signer + deployment packages) to prevent tampered code from running.
TIP
Enable AWS Security Hub with the AWS Foundational Security Best Practices standard — it automatically checks
Lambda for common misconfigurations including public functions, overly permissive roles, and missing tracing.
AWS Lambda Complete Guide | Python Edition | 2025 Page 43
AWS 23 — COST OPTIMISATION & LIMITS AWS Lambda — Python Complete Guide
23 Cost Optimisation & Service Limits
Pricing Model (2025)
Component Price Free Tier (monthly)
Requests $0.20 per million requests 1 million requests
Duration $0.0000166667 per GB-second 400,000 GB-seconds
Provisioned Concurrency $0.0000041667 per GB-second allocated None
Ephemeral Storage (>512MB) $0.0000000309 per GB-second 512 MB included free
Cost Calculation Example
python
# Example: 10 million invocations/month, 512 MB, 200 ms avg duration
requests_cost = (10_000_000 / 1_000_000) * 0.20 # $2.00
gb_seconds = (10_000_000 * 0.2 * 0.512) # 1,024,000 GB-s
duration_cost = (gb_seconds - 400_000) * 0.0000166667 # after free tier
# duration_cost = 624,000 * 0.0000166667 = $10.40
total = requests_cost + duration_cost # $12.40/month
# Compare with 1769 MB (1 vCPU), 80 ms avg — faster but more memory:
gb_seconds_2 = (10_000_000 * 0.08 * 1.769) # 1,415,200 GB-s
duration_cost2 = (gb_seconds_2 - 400_000) * 0.0000166667 # $16.92
# Despite 4x more memory, 2.5x faster -> comparable cost, better UX
Service Limits (Quotas)
Limit Default Adjustable?
Concurrent executions (per region) 1,000 Yes — request increase
Unreserved concurrency minimum 100 No
Function timeout 900 seconds (15 min) No
Memory allocation 128 MB – 10,240 MB No
Deployment package size (zip) 50 MB (direct) / 250 MB unzipped No
Container image size 10 GB No
/tmp ephemeral storage 512 MB – 10,240 MB No (set per function)
Layers per function 5 No
Environment variables 4 KB total No
Function name length 64 characters No
Invocation payload (sync) 6 MB request + 6 MB response No
Invocation payload (async) 256 KB No
SQS batch size 1 – 10,000 messages No
Cost Optimisation Tips
• Use AWS Lambda Power Tuning (open-source) to find optimal memory for cost vs speed.
• Reduce invocation duration: profile with X-Ray, eliminate blocking waits.
• Batch workloads with SQS — process 10 messages per invocation not 1.
• Avoid over-allocating memory: monitor maxMemoryUsed in CloudWatch and right-size.
• Use Graviton2 (arm64) architecture — up to 34% better price/performance than x86.
• Remove dead functions that accumulate CloudWatch log costs.
• Set log retention policies to avoid indefinite log storage costs.
• Use S3 event notifications rather than polling loops.
yaml
# Use Graviton2 (arm64) — significant cost saving
# In SAM template:
AWS Lambda Complete Guide | Python Edition | 2025 Page 44
AWS 23 — COST OPTIMISATION & LIMITS AWS Lambda — Python Complete Guide
Architectures: [arm64]
# In CDK:
architecture=_lambda.Architecture.ARM_64
AWS Lambda Complete Guide | Python Edition | 2025 Page 45
AWS 24 — CHEAT SHEET & QUICK REFERENCE AWS Lambda — Python Complete Guide
24 Complete Cheat Sheet
AWS CLI Lambda Commands
bash
# Function management
aws lambda list-functions
aws lambda get-function --function-name MyFn
aws lambda get-function-configuration --function-name MyFn
aws lambda update-function-code --function-name MyFn --zip-file fileb://[Link]
aws lambda update-function-configuration \
--function-name MyFn --memory-size 512 --timeout 60
# Invoke
aws lambda invoke --function-name MyFn \
--payload '{"key":"val"}' --cli-binary-format raw-in-base64-out [Link]
cat [Link]
# Aliases & Versions
aws lambda publish-version --function-name MyFn
aws lambda create-alias --function-name MyFn --name prod --function-version 5
aws lambda update-alias --function-name MyFn --name prod --function-version 6
# Concurrency
aws lambda put-function-concurrency --function-name MyFn --reserved-concurrent-executions 100
aws lambda put-provisioned-concurrency-config --function-name MyFn --qualifier prod --provisioned-concurr
aws lambda delete-function-concurrency --function-name MyFn
# Layers
aws lambda list-layers
aws lambda list-layer-versions --layer-name my-layer
aws lambda publish-layer-version \
--layer-name my-layer --zip-file fileb://[Link] \
--compatible-runtimes python3.12
# Logs
aws logs tail /aws/lambda/MyFn --follow
boto3 Lambda Client Cheat Sheet
python
import boto3, json
client = [Link]('lambda')
# Synchronous invoke (wait for response)
resp = [Link](
FunctionName='MyFunction',
InvocationType='RequestResponse', # or 'Event' (async) or 'DryRun'
Payload=[Link]({'key': 'value'}),
)
result = [Link](resp['Payload'].read())
# Async invoke (fire and forget)
[Link](
FunctionName='MyFunction',
InvocationType='Event',
Payload=[Link]({'job_id': '123'}),
)
# List functions with pagination
paginator = client.get_paginator('list_functions')
AWS Lambda Complete Guide | Python Edition | 2025 Page 46
AWS 24 — CHEAT SHEET & QUICK REFERENCE AWS Lambda — Python Complete Guide
for page in [Link](MaxItems=50):
for fn in page['Functions']:
print(fn['FunctionName'], fn['Runtime'], fn['MemorySize'])
# Update env vars
client.update_function_configuration(
FunctionName='MyFunction',
Environment={'Variables': {'KEY': 'new_value', 'OTHER': 'x'}},
)
Quick Reference — Lambda Limits & Defaults
Setting Default Max / Range
Memory 128 MB 128 MB – 10,240 MB
Timeout 3 seconds 1 s – 900 s (15 min)
Ephemeral /tmp 512 MB 512 MB – 10,240 MB
Concurrency Unreserved (account pool) 1,000 per region (soft limit)
Layers 0 Max 5 per function
Env vars — Max 4 KB total across all vars
Zip package — 50 MB direct, 250 MB unzipped
Container image — Max 10 GB
Sync payload — 6 MB req / 6 MB resp
Async payload — Max 256 KB
Retries (async) 2 Configurable 0–2
Python Handler Skeleton
python
# lambda_function.py — production-ready skeleton
import json, logging, os, time, boto3
from typing import Any
# Module-level init — runs once, reused on warm invocations
logger = [Link]()
[Link]([Link]('LOG_LEVEL', 'INFO'))
TABLE = [Link]('dynamodb').Table([Link]['TABLE_NAME'])
class AppError(Exception):
def __init__(self, message: str, status: int = 500):
[Link], [Link] = message, status
def handler(event: dict, context: Any) -> dict:
start = [Link]()
[Link]([Link]({
'request_id': context.aws_request_id,
'event_keys': list([Link]()),
}))
try:
result = process(event, context)
[Link]('OK %.0fms', ([Link]() - start) * 1000)
return {'statusCode': 200,
'headers': {'Content-Type': 'application/json'},
'body': [Link](result, default=str)}
except AppError as e:
return {'statusCode': [Link],
'body': [Link]({'error': [Link]})}
except Exception as e:
[Link]('Unhandled error')
AWS Lambda Complete Guide | Python Edition | 2025 Page 47
AWS 24 — CHEAT SHEET & QUICK REFERENCE AWS Lambda — Python Complete Guide
return {'statusCode': 500,
'body': [Link]({'error': 'Internal error'})}
def process(event: dict, context: Any) -> dict:
# Your business logic here
return {'ok': True}
TIP
Official docs: [Link] | SAM: [Link] |
Power Tuning: [Link]/alexcasalboni/aws-lambda-power-tuning
AWS Lambda Complete Guide | Python Edition | 2025 Page 48