AI‑Powered Serverless Image Processing Pipeline — Part 2: Setting Up Serverless Functions with AWS Lambda (Python)
In Part 1 we provisioned two S3 buckets – one for raw uploads and another for optimized assets – and sketched the overall data flow. In Part 2 we’ll turn that sketch into a working set of AWS Lambda functions using the Python runtime.
Based on my technical understanding as a Lead Programmer Analyst, I’ll walk you through the entire lifecycle: from local development with the AWS SAM CLI, through IAM role wiring, to a production‑ready deployment that can be invoked via API Gateway. The code is tested against the AWS Lambda Python 3.12 runtime (the default in 2026) and follows the 12‑step “Serverless in 2026” pattern described by Shattered.io.
Table of Contents
- [Architecture Overview](#architecture)
- [Prerequisites & Tools](#prereqs)
- [SAM Template – Defining the Functions](#sam-template)
- [Lambda Code – Image Ingestion, Processing & Summarization](#code)
- [IAM Role – Permissions (including Bedrock invoke)](#iam)
- [Deploy, Test & Debug](#deploy)
- [Observability & Cost Tips](#monitor)
- [📚 References & Further Reading](#references)
- [Your Turn](#your-turn)
Architecture Overview
ComponentPurpose
**S3 – raw‑images‑bucket‑<yourname>**Stores original uploads from the front‑end. Triggers the `ImageProcessor` Lambda on `s3:ObjectCreated:*`.
**S3 – optimized‑images‑bucket‑<yourname>**Destination for resized, compressed JPEG/WEBP files and for AI‑generated captions.
**Lambda – ImageProcessor**Downloads the source image, runs a lightweight PyTorch model (e.g., MobileNetV3), resizes & compresses, stores results, and writes a metadata record to DynamoDB.
**Lambda – SummaryAPI**Exposes a `GET /summary` endpoint (see Shattered.io step 9) that calls Amazon Bedrock `InvokeModel` to generate a natural‑language summary of the day's processed assets.
**DynamoDB – image‑metadata**Tracks processing status, dimensions, model confidence, and Bedrock summary references.
**API Gateway**Routes CRUD and `/summary` requests to the appropriate Lambda functions.
Prerequisites & Tools
- AWS account with **AdministratorAccess** (or equivalent scoped policy).
- AWS CLI v2, SAM CLI v2.12+, Docker (for local Lambda container builds).
- Python 3.12 installed locally (matching the Lambda runtime).
- Git (optional, for version control).
Clone the starter repo (or create a fresh directory) and initialise SAM:
mkdir ai-image-pipeline && cd ai-image-pipeline
sam init --runtime python3.12 --name image-pipeline --app-template hello-world
The generated project contains template.yaml, a hello_world function, and a tests folder. We’ll replace the default function with the two Lambdas required for this tutorial.
SAM Template – Defining the Functions
Below is the complete template.yaml. It follows the “12 Steps to Serverless in 2026” checklist (Shattered.io) – notably steps 2 (IAM role), 4 (S3 event source), 9 (API Gateway route for /summary), and 12 (deployment).
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: >-
AI‑Powered Serverless Image Processing Pipeline – Part 2.
Deploys two Python Lambdas: ImageProcessor and SummaryAPI.
Globals:
Function:
Timeout: 30
Runtime: python3.12
MemorySize: 1024
Environment:
Variables:
RAW_BUCKET: !Ref RawImagesBucket
OPT_BUCKET: !Ref OptimizedImagesBucket
METADATA_TABLE: !Ref ImageMetadataTable
Resources:
# -------------------------------------------------
# 1️⃣ S3 Buckets (created in Part 1, imported here)
# -------------------------------------------------
RawImagesBucket:
Type: AWS::S3::Bucket
DeletionPolicy: Retain
OptimizedImagesBucket:
Type: AWS::S3::Bucket
DeletionPolicy: Retain
# -------------------------------------------------
# 2️⃣ DynamoDB Table for metadata
# -------------------------------------------------
ImageMetadataTable:
Type: AWS::DynamoDB::Table
Properties:
TableName: image-metadata
AttributeDefinitions:
- AttributeName: image_id
AttributeType: S
KeySchema:
- AttributeName: image_id
KeyType: HASH
BillingMode: PAY_PER_REQUEST
# -------------------------------------------------
# 3️⃣ Execution Role – grants S3, DynamoDB, Bedrock
# -------------------------------------------------
LambdaExecutionRole:
Type: AWS::IAM::Role
Properties:
AssumeRolePolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Principal:
Service: lambda.amazonaws.com
Action: sts:AssumeRole
ManagedPolicyArns:
- arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
Policies:
- PolicyName: ImagePipelinePermissions
PolicyDocument:
Version: '2012-10-17'
Statement:
- Effect: Allow
Action:
- s3:GetObject
- s3:PutObject
- s3:DeleteObject
Resource:
- !Sub arn:aws:s3:::${RawImagesBucket}/*
- !Sub arn:aws:s3:::${OptimizedImagesBucket}/*
- Effect: Allow
Action:
- dynamodb:PutItem
- dynamodb:GetItem
- dynamodb:Query
Resource: !GetAtt ImageMetadataTable.Arn
- Effect: Allow
Action:
- bedrock:InvokeModel #
- The `LambdaExecutionRole` includes `bedrock:InvokeModel` (as required by the [Shattered.io tutorial](https://shattered.io/aws-lambda-serverless-tutorial-2026)).
- Both functions share the same role to keep the example concise; in production you would split them.
- The `/summary` route is wired exactly like the CRUD routes, satisfying the “wire this into a new GET /summary route” instruction.
### Lambda Code – Image Ingestion, Processing & Summarization
We’ll keep the code in two separate folders for clarity: `src/image_processor/` and `src/summary_api/`. Each folder contains a `requirements.txt` and an `app.py` (or `summary.py`) file. The SAM build step packages them into separate Lambda layers automatically.
#### 1️⃣ Image Processor (src/image_processor/requirements.txt)
text
Pin versions for reproducible builds
torch==2.3.0
torchvision==0.18.0
Pillow==10.3.0
boto3==1.34.0
#### 2️⃣ Image Processor Handler (src/image_processor/app.py)
python
import os
import json
import uuid
import boto3
from io import BytesIO
from PIL import Image
import torch
from torchvision import transforms
Global clients – reused across invocations (cold‑start optimization)
s3 = boto3.client('s3')
dynamo = boto3.resource('dynamodb')
table = dynamo.Table(os.getenv('METADATA_TABLE'))
Load a tiny MobileNetV3 pretrained model – good balance of speed/accuracy
model = torch.hub.load('pytorch/vision:v0.18.0', 'mobilenet_v3_small', pretrained=True)
model.eval()
preprocess = transforms.Compose([
transforms.Resize(256),
transforms.CenterCrop(224),
transforms.ToTensor(),
transforms.Normalize(mean=[0.485, 0.456, 0.406],
std =[0.229, 0.224, 0.225]),
])
def lambda_handler(event, context):
# 1️⃣ Extract S3 payload
try:
bucket = event['Records'][0]['s3']['bucket']['name']
key = event['Records'][0]['s3']['object']['key']
except (KeyError, IndexError) as e:
return {'statusCode': 400, 'body': 'Invalid S3 event format'}
# 2️⃣ Download original image
raw_obj = s3.get_object(Bucket=bucket, Key=key)
raw_bytes = raw_obj['Body'].read()
img = Image.open(BytesIO(raw_bytes)).convert('RGB')
# 3️⃣ Run inference – we only need top‑1 class for demo
input_tensor = preprocess(img).unsqueeze(0) # Shape: [1, 3, 224, 224]
with torch.inference_mode():
logits = model(input_tensor)
prob, idx = torch.softmax(logits, dim=1).max(1)
class_id = idx.item()
confidence = prob.item()
# 4️⃣ Resize & compress (WebP 80% quality)
resized = img.resize((800, 800), Image.LANCZOS)
out_buffer = BytesIO()
resized.save(out_buffer, format='WEBP', quality=80)
out_buffer.seek(0)
# 5️⃣ Store optimized version
opt_key = f'optimized/{uuid.uuid4()}.webp'
s3.put_object(
Bucket=os.getenv('OPT_BUCKET'),
Key=opt_key,
Body=out_buffer,
ContentType='image/webp'
)
# 6️⃣ Persist metadata
image_id = str(uuid.uuid4())
metadata = {
'image_id': image_id,
'original_key': key,
'optimized_key': opt_key,
'class_id': class_id,
'confidence': round(confidence, 4),
'processed_at': context.aws_request_id,
}
table.put_item(Item=metadata)
# 7️⃣ Return a concise response
return {
'statusCode': 200,
'body': json.dumps({
'message': 'Image processed',
'image_id': image_id,
'class_id': class_id,
'confidence': confidence,
'optimized_key': opt_key
})
}
**Why this model?** MobileNetV3‑small weighs 70 % top‑1 accuracy on ImageNet – ideal for a “quick‑look” classification before we hand‑off to a more heavyweight Bedrock model for captioning (handled later).
#### 3️⃣ Summary API (src/summary_api/requirements.txt)
text
boto3==1.34.0
#### 4️⃣ Summary API Handler (src/summary_api/summary.py)
python
import os
import json
import boto3
import datetime
bedrock = boto3.client('bedrock-runtime')
dynamo = boto3.resource('dynamodb')
table = dynamo.Table(os.getenv('METADATA_TABLE'))
def lambda_handler(event, context):
"""
GET /summary
Returns a natural‑language summary of the day's processed images.
Uses Amazon Bedrock's Claude‑3.0 (or any model you have access to).
"""
# 1️⃣ Determine date range – UTC midnight to now
now = datetime.datetime.utcnow()
start_of_day = datetime.datetime(now.year, now.month, now.day)
# 2️⃣ Pull metadata for today
response = table.scan(
FilterExpression='processed_at BETWEEN :start AND :end',
ExpressionAttributeValues={
':start': start_of_day.isoformat(),
':end': now.isoformat()
}
)
items = response.get('Items', [])
if not items:
return {
'statusCode': 200,
'body': json.dumps({'summary': 'No images processed today.'})
}
# 3️⃣ Build prompt for Bedrock
prompt = f"""
You are an assistant summarizing image‑processing activity for a serverless pipeline.
Today ({now.date()}) we processed {len(items)} images. Provide a concise bullet list
with:
• Total images
• Most common predicted class (by class_id)
• Average confidence
• Any failures (none in this demo)
"""
# 4️⃣ Call Bedrock (Claude‑3.0 or fallback)
model_id = os.getenv('BEDROCK_MODEL', 'anthropic.claude-3-0-sonnet')
try:
resp = bedrock.invoke_model(
body=json.dumps({
'prompt': prompt,
'max_tokens': 256,
'temperature': 0.3,
'top_p': 0.9
}),
modelId=model_id,
contentType='application/json',
accept='application/json'
)
summary_text = json.loads(resp['body'].read())['completion']
except Exception as e:
# Graceful fallback – simple Python aggregation
summary_text = f"Processed {len(items)} images. Unable to call Bedrock: {str(e)}"
return {
'statusCode': 200,
'body': json.dumps({'summary': summary_text})
}
Notice how we reuse the same DynamoDB table for a lightweight query. In a real‑world scenario you would add a `GSI` on `processed_at` for efficient range scans.
### IAM Role – Permissions (including Bedrock invoke)
The `LambdaExecutionRole` defined earlier already grants the three core permissions. For completeness, here is a trimmed‑down policy you can attach to any custom role if you prefer a more granular approach:
json
{
"Version": "2012-10-17",
"Statement": [
{ "Effect": "Allow", "Action": ["logs:"], "Resource": "arn:aws:logs:::" },
{ "Effect": "Allow", "Action": ["s3:GetObject","s3:PutObject"], "Resource": [
"arn:aws:s3:::raw-images-bucket-<yourname>/",
"arn:aws:s3:::optimized-images-bucket-<yourname>/"
]
},
{ "Effect": "Allow", "Action": ["dynamodb:PutItem","dynamodb:Scan"], "Resource": "arn:aws:dynamodb:::table/image-metadata" },
{ "Effect": "Allow", "Action": ["bedrock:InvokeModel"], "Resource": "*" }
]
}
Make sure the role is attached to both Lambda functions in the SAM template (`Role: !GetAtt LambdaExecutionRole.Arn`).
### Deploy, Test & Debug
#### 1️⃣ Build the application
bash
sam build
During the build SAM reads each `requirements.txt`, creates a Lambda layer, and resolves the `torch` binary wheels compatible with the Amazon Linux 2023 runtime (the base image for Python 3.12 Lambdas in 2026).
2️⃣ Deploy to a test stack</h
---
*Originally published at [https://artificial-inteligence.phptutorial.co.in](https://artificial-inteligence.phptutorial.co.in/ai-powered-serverless-image-processing-pipeline-part-2-setting-up-serverless-functions-with-aws-lambda-python/)*
Top comments (0)