0% found this document useful (0 votes)
3 views49 pages

AWS Lambda Python Guide

The document is a comprehensive guide to AWS Lambda with a focus on Python, covering serverless architecture, core concepts, and practical examples. It includes detailed sections on functions, events, IAM roles, deployment, monitoring, and best practices for using AWS Lambda. The guide is structured to assist users in understanding and implementing AWS Lambda effectively, with a focus on Python 3.12 and associated tools like SAM and CDK.

Uploaded by

ashishk9186
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
3 views49 pages

AWS Lambda Python Guide

The document is a comprehensive guide to AWS Lambda with a focus on Python, covering serverless architecture, core concepts, and practical examples. It includes detailed sections on functions, events, IAM roles, deployment, monitoring, and best practices for using AWS Lambda. The guide is structured to assist users in understanding and implementing AWS Lambda effectively, with a focus on Python 3.12 and associated tools like SAM and CDK.

Uploaded by

ashishk9186
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

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

You might also like