Module 4: LocalStack — AWS Local Development

5. S3 + Lambda Local Pipeline

Next →

Overview

In this capsule you'll connect S3 and Lambda into a complete AI pipeline running on LocalStack. Lambda reads a document from S3, processes it with an LLM, and writes the result back to S3. It's the complete flow: input → processing → output — all local, all free. This pipeline is the central artifact of the module and the basis of the final project (capsule 08).

Context: In the previous capsules you learned local S3 (capsule 03) and local Lambda (capsule 04) separately. Now you combine them. This pattern — storage + compute + LLM — is the building block of most production AI systems. Mastering it locally gives you the confidence to implement it on real AWS (Module 5).

Pipeline Architecture

The complete flow

┌─────────────────────────────────────────────────┐
│                 LocalStack (localhost:4566)       │
│                                                   │
│   ┌──────────────┐     ┌──────────────────────┐  │
│   │  S3 Bucket   │     │    Lambda Function    │  │
│   │  ai-input    │────→│    ai-processor       │  │
│   │              │     │                        │  │
│   │ documents/   │     │  1. Reads from S3      │  │
│   │  doc.txt     │     │  2. Calls OpenAI       │  │
│   └──────────────┘     │  3. Writes to S3       │  │
│                         │                        │  │
│   ┌──────────────┐     └──────────────────────┘  │
│   │  S3 Bucket   │←────────────────────────────  │
│   │  ai-output   │                                │
│   │              │                                │
│   │ results/     │                                │
│   │  doc-result  │                                │
│   └──────────────┘                                │
└─────────────────────────────────────────────────┘

The three components

1. S3 Input (bucket: ai-input)
   └── Stores documents to process (texts, prompts, data)

2. Lambda Processor (function: ai-processor)
   ├── Reads document from S3 (boto3.get_object)
   ├── Processes with an LLM (OpenAI API)
   └── Writes result to S3 (boto3.put_object)

3. S3 Output (bucket: ai-output)
   └── Stores processing results

The Lambda Processor

Complete handler

# lambda/processor.py
import json
import os
import time
import boto3
from openai import OpenAI

AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL", None)

openai_client = OpenAI(
    api_key=os.environ.get("OPENAI_API_KEY", ""),
    timeout=50,
    max_retries=1,
)

MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini")
INPUT_BUCKET = os.environ.get("INPUT_BUCKET", "ai-input")
OUTPUT_BUCKET = os.environ.get("OUTPUT_BUCKET", "ai-output")
SYSTEM_PROMPT = os.environ.get(
    "SYSTEM_PROMPT",
    "Analyze the following document. Extract: title, summary (2-3 sentences), "
    "and a list of key points. Respond in JSON format."
)


def get_s3_client():
    """Creates an S3 client — points to LocalStack if AWS_ENDPOINT_URL is set."""
    kwargs = {
        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
    }
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = "test"
        kwargs["aws_secret_access_key"] = "test"
    return boto3.client("s3", **kwargs)


def handler(event, context):
    start_time = time.time()
    s3 = get_s3_client()

    # 1. Get the key of the document to process
    document_key = _get_document_key(event)
    if not document_key:
        return _response(400, {"error": "document_key is required"})

    # 2. Read the document from S3
    try:
        response = s3.get_object(Bucket=INPUT_BUCKET, Key=document_key)
        document_text = response["Body"].read().decode("utf-8")
    except Exception as e:
        return _response(404, {
            "error": f"Cannot read document: {str(e)}",
            "bucket": INPUT_BUCKET,
            "key": document_key,
        })

    # 3. Process with the LLM
    try:
        llm_response = openai_client.chat.completions.create(
            model=MODEL_NAME,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": document_text},
            ],
            max_tokens=800,
        )
        analysis = llm_response.choices[0].message.content
        tokens_used = llm_response.usage.total_tokens
    except Exception as e:
        return _response(502, {"error": f"LLM processing failed: {str(e)}"})

    # 4. Build the result
    result = {
        "source_document": document_key,
        "analysis": analysis,
        "model": MODEL_NAME,
        "tokens_used": tokens_used,
        "duration_ms": round((time.time() - start_time) * 1000),
        "processed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
    }

    # 5. Write the result to S3
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    try:
        s3.put_object(
            Bucket=OUTPUT_BUCKET,
            Key=output_key,
            Body=json.dumps(result, indent=2, ensure_ascii=False),
            ContentType="application/json",
        )
    except Exception as e:
        return _response(500, {"error": f"Cannot write result: {str(e)}"})

    return _response(200, {
        "status": "processed",
        "input": f"s3://{INPUT_BUCKET}/{document_key}",
        "output": f"s3://{OUTPUT_BUCKET}/{output_key}",
        "tokens_used": tokens_used,
        "duration_ms": result["duration_ms"],
    })


def _get_document_key(event):
    """Extracts document_key from the event — supports direct invocation and API Gateway."""
    if "body" in event:
        try:
            body = json.loads(event["body"])
            return body.get("document_key", "")
        except (json.JSONDecodeError, TypeError):
            pass
    return event.get("document_key", "")


def _response(status_code, body):
    return {
        "statusCode": status_code,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps(body, ensure_ascii=False),
    }

The important parts of the handler

There are three key design decisions:

1. AWS_ENDPOINT_URL controls the target
   ├── If defined → uses LocalStack
   └── If not → uses real AWS (boto3 default)
   This is the environment switching mechanism (capsule 06)

2. The handler reads from S3 and writes to S3
   ├── Input: get_object from the ai-input bucket
   ├── Process: OpenAI API
   └── Output: put_object to the ai-output bucket
   S3 is the data bus between components

3. The result includes metadata
   ├── source_document, model, tokens, duration
   └── Useful for auditing, debugging, and cost tracking

Configure the Pipeline

Create the buckets

# Create the input and output buckets
awslocal s3 mb s3://ai-input
awslocal s3 mb s3://ai-output

# Verify
awslocal s3 ls
# ai-input
# ai-output

Upload a test document

# Create a text document
cat > data/test-document.txt << 'EOF'
Q1 2026 Quarterly Report — Technology Department

Executive Summary:
The technology department completed the migration to a microservices architecture
during Q1 2026. Deploy times were reduced from 4 hours to 15 minutes.
The AI system for ticket classification reduced resolution time
by 40%.

Main achievements:
- Complete migration to Kubernetes (EKS)
- Automated CI/CD pipeline with GitHub Actions
- Ticket classification system with GPT-4o-mini
- Cloud cost reduction of 25% via right-sizing

Next steps:
- Implement monitoring with Prometheus + Grafana
- Expand the AI system to customer support
- Evaluate migrating the database to Aurora Serverless
EOF

# Upload to S3
awslocal s3 cp data/test-document.txt s3://ai-input/documents/report-q1.txt

# Verify it's in S3
awslocal s3 ls s3://ai-input/documents/

Package and deploy the Lambda processor

# Install dependencies
pip install openai boto3 -t lambda/package/ --quiet

# Copy the handler
cp lambda/processor.py lambda/package/

# Package
cd lambda/package
zip -r ../processor.zip . -q
cd ../..

# Deploy
awslocal lambda create-function \
  --function-name ai-processor \
  --runtime python3.11 \
  --handler processor.handler \
  --zip-file fileb://lambda/processor.zip \
  --role arn:aws:iam::000000000000:role/lambda-role \
  --timeout 90 \
  --memory-size 768 \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    INPUT_BUCKET=ai-input,
    OUTPUT_BUCKET=ai-output,
    AWS_ENDPOINT_URL=http://host.docker.internal:4566,
    AWS_DEFAULT_REGION=us-east-1
  }"

The variable AWS_ENDPOINT_URL=http://host.docker.internal:4566 is necessary because Lambda in LocalStack runs in a separate Docker container. host.docker.internal resolves to the host from inside the container.

If you use LAMBDA_EXECUTOR=local, the endpoint would be http://localhost:4566.

Run the Pipeline

Manual invocation

# Invoke the processor with the document we uploaded
awslocal lambda invoke \
  --function-name ai-processor \
  --payload '{"document_key": "documents/report-q1.txt"}' \
  --cli-binary-format raw-in-base64-out \
  output.json

# See the invocation result
cat output.json | python3 -m json.tool

Verify the result in S3

# List the results
awslocal s3 ls s3://ai-output/results/

# Download and see the analysis
awslocal s3 cp s3://ai-output/results/report-q1-txt-analysis.json - | python3 -m json.tool

Run from Python

# run_pipeline.py
import boto3
import json

ENDPOINT = "http://localhost:4566"
CLIENT_KWARGS = {
    "endpoint_url": ENDPOINT,
    "aws_access_key_id": "test",
    "aws_secret_access_key": "test",
    "region_name": "us-east-1",
}

s3 = boto3.client("s3", **CLIENT_KWARGS)
lambda_client = boto3.client("lambda", **CLIENT_KWARGS)


def run_pipeline(document_key):
    """Runs the complete pipeline: S3 → Lambda → S3."""
    print(f"Processing: {document_key}")

    # Invoke Lambda
    response = lambda_client.invoke(
        FunctionName="ai-processor",
        InvocationType="RequestResponse",
        Payload=json.dumps({"document_key": document_key}),
    )

    result = json.loads(response["Payload"].read())
    body = json.loads(result["body"])

    if result["statusCode"] == 200:
        print(f"  Input:  {body['input']}")
        print(f"  Output: {body['output']}")
        print(f"  Tokens: {body['tokens_used']}")
        print(f"  Duration: {body['duration_ms']}ms")

        # Download the result
        output_key = body["output"].split(f"ai-output/")[1]
        obj = s3.get_object(Bucket="ai-output", Key=output_key)
        analysis = json.loads(obj["Body"].read().decode("utf-8"))
        print(f"  Analysis: {analysis['analysis'][:150]}...")
    else:
        print(f"  ERROR: {body}")

    return body


# Run
result = run_pipeline("documents/report-q1.txt")

Batch Processing: Multiple Documents

Upload several documents

# batch_upload.py
import boto3

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

documents = {
    "documents/customer-email.txt": (
        "Subject: Billing problem\n\n"
        "Dear team, I've had an incorrect charge on my account for 3 months. "
        "The Premium service should cost $49/month but I'm charged $79. "
        "I've contacted support twice with no resolution. I need you to correct "
        "the charge and refund me the difference. Thanks, Maria Gonzalez."
    ),
    "documents/technical-proposal.txt": (
        "Proposal: Recommendation System for E-commerce\n\n"
        "Objective: Implement a recommendation system based on "
        "embeddings that improves the CTR by 20%.\n"
        "Technology: FastAPI + Redis + OpenAI Embeddings\n"
        "Timeline: 6 weeks\n"
        "Budget: $15,000 USD\n"
        "Team: 2 ML Engineers + 1 Backend Developer"
    ),
    "documents/bug-report.txt": (
        "Bug Report #4521\n"
        "Severity: High\n"
        "Component: Payments API\n"
        "Description: When processing payments with international cards, "
        "the system returns a 500 error intermittently. It affects 15% "
        "of international transactions. Logs show a timeout "
        "on the connection with the payment gateway."
    ),
}

for key, content in documents.items():
    s3.put_object(Bucket="ai-input", Key=key, Body=content.encode("utf-8"))
    print(f"Uploaded: {key}")

Process the batch

# batch_process.py
import boto3
import json
import time

ENDPOINT = "http://localhost:4566"
KWARGS = {
    "endpoint_url": ENDPOINT,
    "aws_access_key_id": "test",
    "aws_secret_access_key": "test",
    "region_name": "us-east-1",
}

s3 = boto3.client("s3", **KWARGS)
lambda_client = boto3.client("lambda", **KWARGS)


def process_all_documents():
    """Processes all documents in ai-input/documents/."""
    response = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
    documents = [obj["Key"] for obj in response.get("Contents", [])]

    print(f"Documents found: {len(documents)}")
    results = []

    for doc_key in documents:
        print(f"\nProcessing: {doc_key}")
        start = time.time()

        resp = lambda_client.invoke(
            FunctionName="ai-processor",
            InvocationType="RequestResponse",
            Payload=json.dumps({"document_key": doc_key}),
        )

        result = json.loads(resp["Payload"].read())
        body = json.loads(result["body"])
        elapsed = round((time.time() - start) * 1000)

        status = "OK" if result["statusCode"] == 200 else "ERROR"
        tokens = body.get("tokens_used", 0)
        print(f"  {status} — {tokens} tokens — {elapsed}ms")

        results.append({
            "document": doc_key,
            "status": status,
            "tokens": tokens,
            "elapsed_ms": elapsed,
        })

    # Summary
    print("\n" + "=" * 50)
    total_tokens = sum(r["tokens"] for r in results)
    total_time = sum(r["elapsed_ms"] for r in results)
    success = sum(1 for r in results if r["status"] == "OK")
    print(f"Processed: {success}/{len(results)}")
    print(f"Total tokens: {total_tokens}")
    print(f"Total time: {total_time}ms")
    print(f"Estimated cost (gpt-4o-mini): ~${total_tokens * 0.0000015:.4f}")


process_all_documents()

Verify Results

List all results

# See all the generated analyses
awslocal s3 ls s3://ai-output/results/ --recursive

# Download a specific one
awslocal s3 cp s3://ai-output/results/customer-email-txt-analysis.json - | python3 -m json.tool

Verification script

# verify_results.py
import boto3
import json

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

# List inputs and outputs
inputs = s3.list_objects_v2(Bucket="ai-input", Prefix="documents/")
outputs = s3.list_objects_v2(Bucket="ai-output", Prefix="results/")

input_keys = [o["Key"] for o in inputs.get("Contents", [])]
output_keys = [o["Key"] for o in outputs.get("Contents", [])]

print(f"Input documents: {len(input_keys)}")
print(f"Generated results:  {len(output_keys)}")

for key in output_keys:
    obj = s3.get_object(Bucket="ai-output", Key=key)
    result = json.loads(obj["Body"].read().decode("utf-8"))
    print(f"\n--- {key} ---")
    print(f"  Source:   {result['source_document']}")
    print(f"  Model:    {result['model']}")
    print(f"  Tokens:   {result['tokens_used']}")
    print(f"  Duration: {result['duration_ms']}ms")
    print(f"  Analysis: {result['analysis'][:120]}...")

Exercises

Exercise 1: Pipeline with a system prompt from S3

Modify the processor so it reads the system prompt from a file in S3 (ai-config/prompts/analyzer.txt) instead of having it hardcoded. If the file doesn't exist, use the default prompt.

See solution 
# In the handler, add a function to read the prompt from S3:
CONFIG_BUCKET = os.environ.get("CONFIG_BUCKET", "ai-config")
PROMPT_KEY = os.environ.get("PROMPT_KEY", "prompts/analyzer.txt")

def get_system_prompt(s3):
    """Reads the system prompt from S3, with a fallback to the default."""
    default = (
        "Analyze the following document. Extract: title, summary, "
        "and key points. Respond in JSON."
    )
    try:
        response = s3.get_object(Bucket=CONFIG_BUCKET, Key=PROMPT_KEY)
        return response["Body"].read().decode("utf-8")
    except Exception:
        return default

# In handler(), replace the SYSTEM_PROMPT constant:
def handler(event, context):
    s3 = get_s3_client()
    system_prompt = get_system_prompt(s3)
    # ... use system_prompt in the OpenAI call
# Setup: create the config bucket and upload the prompt
awslocal s3 mb s3://ai-config
echo "You are a document analyst. Extract: document category, urgency level (high/medium/low), and the 3 required actions. Respond in JSON." | \
  awslocal s3 cp - s3://ai-config/prompts/analyzer.txt

# Redeploy and test

Exercise 2: Pipeline with processing tracking

Add a tracking file to the pipeline: each time a document is processed, append a line to ai-output/tracking/log.jsonl (JSON Lines) with the document key, timestamp, tokens, and status.

See solution 
import datetime

def append_to_tracking(s3, document_key, tokens, status, duration_ms):
    """Appends an entry to the tracking log in S3."""
    tracking_key = "tracking/log.jsonl"
    
    entry = json.dumps({
        "document": document_key,
        "timestamp": datetime.datetime.utcnow().isoformat() + "Z",
        "tokens": tokens,
        "status": status,
        "duration_ms": duration_ms,
    })

    # Read the existing log (if any)
    try:
        existing = s3.get_object(Bucket=OUTPUT_BUCKET, Key=tracking_key)
        current_log = existing["Body"].read().decode("utf-8")
    except Exception:
        current_log = ""

    # Append the new entry
    updated_log = current_log + entry + "\n"

    s3.put_object(
        Bucket=OUTPUT_BUCKET,
        Key=tracking_key,
        Body=updated_log.encode("utf-8"),
        ContentType="application/x-ndjson",
    )

# Call at the end of the handler:
# append_to_tracking(s3, document_key, tokens_used, "success", duration_ms)
# After processing several documents:
awslocal s3 cp s3://ai-output/tracking/log.jsonl -
# {"document": "documents/report-q1.txt", "timestamp": "...", "tokens": 342, ...}
# {"document": "documents/customer-email.txt", "timestamp": "...", "tokens": 215, ...}

Exercise 3: Pipeline with duplicate-document checking

Before processing a document, check if a result already exists in ai-output. If it was already processed, return the existing result without reprocessing (avoiding unnecessary token spend).

See solution 
def check_existing_result(s3, document_key):
    """Checks if a document has already been processed."""
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}-analysis.json"

    try:
        response = s3.get_object(Bucket=OUTPUT_BUCKET, Key=output_key)
        existing = json.loads(response["Body"].read().decode("utf-8"))
        return output_key, existing
    except Exception:
        return None, None

# In handler(), before calling the LLM:
def handler(event, context):
    s3 = get_s3_client()
    document_key = _get_document_key(event)

    # Check if a result already exists
    force = False
    if "body" in event:
        try:
            body = json.loads(event["body"])
            force = body.get("force", False)
        except Exception:
            pass

    if not force:
        existing_key, existing_result = check_existing_result(s3, document_key)
        if existing_result:
            return _response(200, {
                "status": "already_processed",
                "output": f"s3://{OUTPUT_BUCKET}/{existing_key}",
                "tokens_used": existing_result.get("tokens_used", 0),
                "message": "Existing result returned (use force=true to reprocess)",
            })

    # ... continue with normal processing
# First invocation: processes
awslocal lambda invoke --function-name ai-processor \
  --payload '{"document_key": "documents/report-q1.txt"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "processed"

# Second invocation: returns the existing one
awslocal lambda invoke --function-name ai-processor \
  --payload '{"document_key": "documents/report-q1.txt"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "already_processed"

# Force reprocessing:
awslocal lambda invoke --function-name ai-processor \
  --payload '{"body": "{\"document_key\": \"documents/report-q1.txt\", \"force\": true}"}' \
  --cli-binary-format raw-in-base64-out output.json
# status: "processed"

Exercise 4: End-to-end pipeline script

Create a bash script that runs the complete pipeline: creates buckets, uploads a document, deploys Lambda, invokes, downloads the result, and shows the analysis. A single command for a demo.

See solution 
#!/bin/bash
# scripts/demo-pipeline.sh
set -e

echo "=== Pipeline S3 + Lambda Demo ==="
echo ""

echo "1. Verifying LocalStack..."
curl -s http://localhost:4566/_localstack/health > /dev/null || { echo "LocalStack not available"; exit 1; }
echo "   OK"

echo "2. Creating buckets..."
awslocal s3 mb s3://ai-input 2>/dev/null || true
awslocal s3 mb s3://ai-output 2>/dev/null || true
echo "   Buckets: ai-input, ai-output"

echo "3. Uploading a test document..."
cat > /tmp/demo-doc.txt << 'CONTENT'
Project Proposal: Chatbot for Technical Support

The team proposes implementing a chatbot based on GPT-4o-mini to automate
60% of technical support inquiries. The chatbot will use RAG over the
existing documentation. Budget: $8,000. Timeline: 4 weeks.
CONTENT
awslocal s3 cp /tmp/demo-doc.txt s3://ai-input/documents/demo.txt
echo "   Document uploaded"

echo "4. Deploying the Lambda processor..."
if awslocal lambda get-function --function-name ai-processor > /dev/null 2>&1; then
  awslocal lambda update-function-code \
    --function-name ai-processor \
    --zip-file fileb://lambda/processor.zip > /dev/null
else
  awslocal lambda create-function \
    --function-name ai-processor \
    --runtime python3.11 \
    --handler processor.handler \
    --zip-file fileb://lambda/processor.zip \
    --role arn:aws:iam::000000000000:role/lambda-role \
    --timeout 90 --memory-size 768 \
    --environment "Variables={OPENAI_API_KEY=${OPENAI_API_KEY},MODEL_NAME=gpt-4o-mini,INPUT_BUCKET=ai-input,OUTPUT_BUCKET=ai-output,AWS_ENDPOINT_URL=http://host.docker.internal:4566}" > /dev/null
fi
echo "   Lambda deployed"

echo "5. Running the pipeline..."
awslocal lambda invoke \
  --function-name ai-processor \
  --payload '{"document_key": "documents/demo.txt"}' \
  --cli-binary-format raw-in-base64-out \
  /tmp/pipeline-result.json > /dev/null

STATUS=$(cat /tmp/pipeline-result.json | python3 -c "import json,sys; print(json.load(sys.stdin)['statusCode'])")

if [ "$STATUS" = "200" ]; then
  echo "   Pipeline ran successfully"
  
  echo ""
  echo "6. Analysis result:"
  echo "   ========================"
  awslocal s3 cp s3://ai-output/results/demo-txt-analysis.json - 2>/dev/null | \
    python3 -c "
import json, sys
data = json.load(sys.stdin)
print(f\"   Model: {data['model']}\")
print(f\"   Tokens: {data['tokens_used']}\")
print(f\"   Duration: {data['duration_ms']}ms\")
print(f\"   Analysis:\")
print(f\"   {data['analysis'][:300]}...\")
"
else
  echo "   ERROR — Status: $STATUS"
  cat /tmp/pipeline-result.json | python3 -m json.tool
fi

echo ""
echo "=== Pipeline complete ==="

Troubleshooting

"Lambda can't connect to S3 in LocalStack"

# If LAMBDA_EXECUTOR=docker, Lambda runs in a separate container
# It needs to use host.docker.internal to reach LocalStack

# Check the configured endpoint:
awslocal lambda get-function-configuration \
  --function-name ai-processor \
  --query 'Environment.Variables.AWS_ENDPOINT_URL'

# It should be: http://host.docker.internal:4566
# NOT: http://localhost:4566 (localhost inside the Lambda container ≠ your machine)

# If you use LAMBDA_EXECUTOR=local, use http://localhost:4566

"The result in S3 is empty or corrupt"

# Verify the body is serialized as a JSON string:
s3.put_object(
    Bucket=bucket,
    Key=key,
    Body=json.dumps(result, ensure_ascii=False),  # string, not dict
    ContentType="application/json",
)

"Timeout when invoking Lambda"

# The default Lambda timeout in LocalStack is 3s
# For AI workloads you need more:
awslocal lambda update-function-configuration \
  --function-name ai-processor \
  --timeout 120

# The Lambda client timeout matters too:
# boto3 has a 60s connection timeout by default

"Lambda processes but the result doesn't appear in S3"

# Verify the buckets exist:
awslocal s3 ls

# Verify there's no silent error in the handler
# Add logging:
import logging
logger = logging.getLogger()
logger.setLevel("DEBUG")

Summary

  • The S3 → Lambda → S3 pipeline is the building block of AI systems in the cloud: input, LLM processing, output.
  • AWS_ENDPOINT_URL in the handler controls whether Lambda talks to LocalStack or AWS — without changing a line of code.
  • Batch processing applies the pipeline to multiple documents sequentially, with tracking of tokens and costs.
  • Deduplication avoids reprocessing already-analyzed documents — saves tokens and money.
  • The S3 endpoint from Lambda depends on the LAMBDA_EXECUTOR: host.docker.internal for Docker, localhost for local.
  • This whole pipeline runs locally, for free — the same flow as on AWS, but at no cost.

Additional Resources

  1. S3 Event Notifications — Trigger Lambda automatically when a file is uploaded to S3
  2. Lambda + S3 Tutorial (AWS) — Official Lambda with S3 tutorial
  3. LocalStack S3 + Lambda — S3 on LocalStack
  4. boto3 S3 Transfers — Efficient upload/download
  5. JSON Lines Format — Structured logging format
  6. Lambda Environment Variables — Variable management
← Previous4. Lambda Locally with LocalStackNext →6. Environment Switching