Module 4: LocalStack — AWS Local Development

3. Local S3 for AI

Next →

Overview

In this capsule you'll master S3 on LocalStack with boto3: create buckets, upload and download files, list objects, and generate presigned URLs. Every example is AI-specific: you store embeddings, prompts, model configurations, and inference results. It's not a generic S3 tutorial — it's S3 as a storage layer for your AI pipeline.

Context: In the previous capsule you configured LocalStack in your Docker Compose. Now you have a working local S3. Here you learn to operate it with Python. The patterns you implement here are reused in capsule 05 (S3 + Lambda pipeline) and in the final project (capsule 08). Everything you do against LocalStack works identically against real AWS — only the endpoint URL changes.

S3 Fundamentals for AI

What S3 is (in an AI context)

S3 (Simple Storage Service) is AWS's object storage service. For AI systems, S3 is where you store:

  • Input data: Documents to process, images to analyze, audio to transcribe
  • Configurations: System prompts, model parameters, response templates
  • Results: LLM responses, generated embeddings, completed analyses
  • Model assets: Model weights (if you use your own models), embedding vectors
Typical AI pipeline with S3:
    ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
    │  S3 Input    │ ──→ │  Processor   │ ──→ │  S3 Output   │
    │              │     │  (Lambda +   │     │              │
    │ documents/   │     │   LLM API)   │     │ results/     │
    │ prompts/     │     │              │     │ analyses/    │
    │ config/      │     │              │     │ embeddings/  │
    └──────────────┘     └──────────────┘     └──────────────┘

Key concepts

Bucket:
├── Main container of objects (like a root directory)
├── The name must be unique (global in AWS, local in LocalStack)
├── Example: "ai-input", "ai-output", "ai-models"

Object:
├── A file stored in a bucket
├── Identified by a Key (path within the bucket)
├── Example: Key = "prompts/system-v2.txt"
├── There are no real folders — the "/" in the key is a convention

Key:
├── The object's "path" within the bucket
├── Example: "documents/invoice-001.pdf"
├── Prefixes simulate folders: "documents/", "results/"

Connect boto3 to LocalStack

S3 client

import boto3
import json

def get_s3_client(endpoint_url="http://localhost:4566"):
    """Creates an S3 client for LocalStack."""
    return boto3.client(
        "s3",
        endpoint_url=endpoint_url,
        aws_access_key_id="test",
        aws_secret_access_key="test",
        region_name="us-east-1",
    )

s3 = get_s3_client()

This pattern — a function that accepts endpoint_url as a parameter — is the basis of the environment switching you'll see in capsule 06. You change the URL and your code points to real AWS.

Resource vs Client

boto3 has two interfaces: client (low-level) and resource (high-level). For AI operations, client is more explicit and predictable:

# Client (recommended — explicit, predictable)
s3_client = boto3.client("s3", endpoint_url="http://localhost:4566")
s3_client.put_object(Bucket="my-bucket", Key="file.json", Body=data)

# Resource (high-level — more "pythonic" but less control)
s3_resource = boto3.resource("s3", endpoint_url="http://localhost:4566")
s3_resource.Bucket("my-bucket").put_object(Key="file.json", Body=data)

In this guide we use client because it's what you'll see in production code and in the AWS documentation.

Bucket Operations

Create buckets

s3 = get_s3_client()

# Create buckets for your AI pipeline
buckets = ["ai-input", "ai-output", "ai-config"]

for bucket_name in buckets:
    try:
        s3.create_bucket(Bucket=bucket_name)
        print(f"Bucket '{bucket_name}' created")
    except s3.exceptions.BucketAlreadyOwnedByYou:
        print(f"Bucket '{bucket_name}' already exists")

List buckets

response = s3.list_buckets()

print("Available buckets:")
for bucket in response["Buckets"]:
    print(f"  - {bucket['Name']} (created: {bucket['CreationDate']})")

Delete bucket

# A bucket must be empty to delete it
# First delete all the objects
def empty_and_delete_bucket(s3, bucket_name):
    """Empties and deletes a bucket."""
    objects = s3.list_objects_v2(Bucket=bucket_name)
    if "Contents" in objects:
        for obj in objects["Contents"]:
            s3.delete_object(Bucket=bucket_name, Key=obj["Key"])
            print(f"  Deleted: {obj['Key']}")

    s3.delete_bucket(Bucket=bucket_name)
    print(f"Bucket '{bucket_name}' deleted")

Upload Files

Upload text (prompts, configs)

# Upload a system prompt
system_prompt = """You are an assistant specialized in document analysis.
Your task is to extract key information from business documents.
Always respond in JSON format with the fields: title, summary, key_points."""

s3.put_object(
    Bucket="ai-config",
    Key="prompts/system-prompt-v2.txt",
    Body=system_prompt.encode("utf-8"),
    ContentType="text/plain",
)
print("System prompt uploaded to S3")

Upload JSON (model configurations, results)

# Upload a model configuration
model_config = {
    "model": "gpt-4o-mini",
    "max_tokens": 500,
    "temperature": 0.3,
    "system_prompt_key": "prompts/system-prompt-v2.txt",
    "version": "2.0",
    "created": "2026-03-08",
}

s3.put_object(
    Bucket="ai-config",
    Key="models/analyzer-config.json",
    Body=json.dumps(model_config, indent=2),
    ContentType="application/json",
)
print("Model configuration uploaded")

Upload binary files (documents to process)

# Upload a document from disk
with open("data/invoice-example.pdf", "rb") as f:
    s3.put_object(
        Bucket="ai-input",
        Key="documents/invoice-001.pdf",
        Body=f.read(),
        ContentType="application/pdf",
    )
print("Document uploaded for processing")

# Alternative: upload_file (more efficient for large files)
s3.upload_file(
    Filename="data/invoice-example.pdf",
    Bucket="ai-input",
    Key="documents/invoice-002.pdf",
)

Upload inference results

# Result of an AI analysis
analysis_result = {
    "document_key": "documents/invoice-001.pdf",
    "model": "gpt-4o-mini",
    "analysis": {
        "title": "Invoice #12345",
        "summary": "Consulting services invoice for $5,000 USD",
        "key_points": [
            "Supplier: Acme Corp",
            "Date: 2026-03-01",
            "Total: $5,000 USD",
            "Item: Technical consulting",
        ],
    },
    "tokens_used": 342,
    "duration_ms": 2100,
    "processed_at": "2026-03-08T15:30:00Z",
}

s3.put_object(
    Bucket="ai-output",
    Key="results/invoice-001-analysis.json",
    Body=json.dumps(analysis_result, indent=2, ensure_ascii=False),
    ContentType="application/json",
)
print("Analysis result saved to S3")

Download Files

Download as a string

# Download and read a text file
response = s3.get_object(
    Bucket="ai-config",
    Key="prompts/system-prompt-v2.txt",
)
system_prompt = response["Body"].read().decode("utf-8")
print(f"System prompt:\n{system_prompt}")

Download JSON

# Download and parse JSON
response = s3.get_object(
    Bucket="ai-output",
    Key="results/invoice-001-analysis.json",
)
result = json.loads(response["Body"].read().decode("utf-8"))
print(f"Title: {result['analysis']['title']}")
print(f"Summary: {result['analysis']['summary']}")

Download to a local file

# Download to disk
s3.download_file(
    Bucket="ai-output",
    Key="results/invoice-001-analysis.json",
    Filename="data/output/invoice-001-analysis.json",
)
print("File downloaded to data/output/")

Check if an object exists

def object_exists(s3, bucket, key):
    """Checks if an object exists in S3."""
    try:
        s3.head_object(Bucket=bucket, Key=key)
        return True
    except s3.exceptions.ClientError:
        return False

exists = object_exists(s3, "ai-output", "results/invoice-001-analysis.json")
print(f"Result exists: {exists}")

List Objects

List all objects in a bucket

response = s3.list_objects_v2(Bucket="ai-input")

if "Contents" in response:
    print(f"Objects in ai-input ({response['KeyCount']}):")
    for obj in response["Contents"]:
        size_kb = obj["Size"] / 1024
        print(f"  {obj['Key']} ({size_kb:.1f} KB)")
else:
    print("Empty bucket")

List by prefix (simulate folders)

# Only objects in "documents/"
response = s3.list_objects_v2(
    Bucket="ai-input",
    Prefix="documents/",
)

print("Pending documents:")
for obj in response.get("Contents", []):
    print(f"  {obj['Key']}")

# Only objects in "results/"
response = s3.list_objects_v2(
    Bucket="ai-output",
    Prefix="results/",
)

print("Processed results:")
for obj in response.get("Contents", []):
    print(f"  {obj['Key']}")

Pagination for buckets with many objects

def list_all_objects(s3, bucket, prefix=""):
    """Lists all objects, handling pagination."""
    paginator = s3.get_paginator("list_objects_v2")
    pages = paginator.paginate(Bucket=bucket, Prefix=prefix)

    all_objects = []
    for page in pages:
        for obj in page.get("Contents", []):
            all_objects.append(obj)

    return all_objects

objects = list_all_objects(s3, "ai-output", prefix="results/")
print(f"Total results: {len(objects)}")

Presigned URLs

What they are and what they're for

Presigned URLs let you give temporary access to an S3 object without exposing credentials. They're useful when your AI app needs to:

  • Let a user download a result without authenticating with AWS
  • Generate a temporary link to share an analysis
  • Allow direct upload from a browser

Generate a download URL

# Generate a temporary URL to download a result
url = s3.generate_presigned_url(
    "get_object",
    Params={
        "Bucket": "ai-output",
        "Key": "results/invoice-001-analysis.json",
    },
    ExpiresIn=3600,  # 1 hour
)

print(f"Download URL (expires in 1h):\n{url}")
# In LocalStack: http://localhost:4566/ai-output/results/...?X-Amz-...

Generate an upload URL

# Generate a temporary URL to upload a file
upload_url = s3.generate_presigned_url(
    "put_object",
    Params={
        "Bucket": "ai-input",
        "Key": "documents/new-document.pdf",
        "ContentType": "application/pdf",
    },
    ExpiresIn=900,  # 15 minutes
)

print(f"Upload URL (expires in 15min):\n{upload_url}")

Use a presigned URL with requests

import requests

# Download using a presigned URL
download_url = s3.generate_presigned_url(
    "get_object",
    Params={"Bucket": "ai-output", "Key": "results/invoice-001-analysis.json"},
    ExpiresIn=3600,
)

response = requests.get(download_url)
result = response.json()
print(f"Downloaded via presigned URL: {result['analysis']['title']}")

AI Patterns with S3

Pattern 1: Config Store — load a model configuration from S3

def load_model_config(s3, bucket="ai-config", key="models/analyzer-config.json"):
    """Loads a model configuration from S3."""
    response = s3.get_object(Bucket=bucket, Key=key)
    return json.loads(response["Body"].read().decode("utf-8"))

config = load_model_config(s3)
print(f"Model: {config['model']}, Temp: {config['temperature']}")

Pattern 2: Result Store — save and retrieve inference results

import datetime

def save_inference_result(s3, document_key, result, bucket="ai-output"):
    """Saves the inference result to S3."""
    timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    doc_name = document_key.split("/")[-1].replace(".", "-")
    output_key = f"results/{doc_name}/{timestamp}.json"

    s3.put_object(
        Bucket=bucket,
        Key=output_key,
        Body=json.dumps(result, indent=2, ensure_ascii=False),
        ContentType="application/json",
    )
    return output_key

output_key = save_inference_result(s3, "documents/invoice-001.pdf", analysis_result)
print(f"Result saved to: {output_key}")

Pattern 3: Batch Processing — process multiple documents

def get_pending_documents(s3, bucket="ai-input", prefix="documents/"):
    """Lists documents pending processing."""
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    return [obj["Key"] for obj in response.get("Contents", [])]

def is_processed(s3, document_key, output_bucket="ai-output"):
    """Checks if a document has already been processed."""
    doc_name = document_key.split("/")[-1].replace(".", "-")
    prefix = f"results/{doc_name}/"
    response = s3.list_objects_v2(Bucket=output_bucket, Prefix=prefix)
    return response.get("KeyCount", 0) > 0

pending = get_pending_documents(s3)
for doc_key in pending:
    if not is_processed(s3, doc_key):
        print(f"Pending: {doc_key}")
    else:
        print(f"Already processed: {doc_key}")

Exercises

Exercise 1: Create a prompt repository in S3

Create a prompt-library bucket and upload 3 different system prompts (analyzer, summarizer, translator). Then write a function that receives the prompt name and downloads it from S3.

See solution 
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",
)

s3.create_bucket(Bucket="prompt-library")

prompts = {
    "analyzer": "Analyze the following document and extract: title, summary, key points. Respond in JSON.",
    "summarizer": "Summarize the following text in at most 3 sentences. Be concise and precise.",
    "translator": "Translate the following text to English. Keep the original tone and format.",
}

for name, content in prompts.items():
    s3.put_object(
        Bucket="prompt-library",
        Key=f"system/{name}.txt",
        Body=content.encode("utf-8"),
        ContentType="text/plain",
    )
    print(f"Prompt '{name}' uploaded")


def get_prompt(s3, prompt_name, bucket="prompt-library"):
    """Downloads a system prompt by name."""
    key = f"system/{prompt_name}.txt"
    try:
        response = s3.get_object(Bucket=bucket, Key=key)
        return response["Body"].read().decode("utf-8")
    except s3.exceptions.NoSuchKey:
        raise ValueError(f"Prompt '{prompt_name}' not found")


for name in ["analyzer", "summarizer", "translator"]:
    prompt = get_prompt(s3, name)
    print(f"\n{name}: {prompt[:60]}...")

Exercise 2: Result versioning

Implement a system that saves multiple versions of an analysis result for the same document. Each run of the analysis creates a new file with a timestamp. Write a function that lists all the versions and another that downloads the most recent one.

See solution 
import boto3
import json
import datetime

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

s3.create_bucket(Bucket="ai-versions")


def save_versioned_result(s3, doc_id, result, bucket="ai-versions"):
    """Saves a result with a timestamp as the version."""
    timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
    key = f"results/{doc_id}/{timestamp}.json"
    result["version_timestamp"] = timestamp

    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=json.dumps(result, indent=2),
        ContentType="application/json",
    )
    return key


def list_versions(s3, doc_id, bucket="ai-versions"):
    """Lists all the versions of a result."""
    prefix = f"results/{doc_id}/"
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    versions = []
    for obj in response.get("Contents", []):
        versions.append({
            "key": obj["Key"],
            "timestamp": obj["Key"].split("/")[-1].replace(".json", ""),
            "size": obj["Size"],
        })
    return sorted(versions, key=lambda x: x["timestamp"], reverse=True)


def get_latest_version(s3, doc_id, bucket="ai-versions"):
    """Downloads the most recent version."""
    versions = list_versions(s3, doc_id, bucket)
    if not versions:
        return None
    response = s3.get_object(Bucket=bucket, Key=versions[0]["key"])
    return json.loads(response["Body"].read().decode("utf-8"))


# Simulate 3 runs
import time

for i in range(3):
    result = {"doc_id": "invoice-001", "score": 0.85 + i * 0.05, "run": i + 1}
    key = save_versioned_result(s3, "invoice-001", result)
    print(f"Version {i+1} saved: {key}")
    time.sleep(1)

print("\nAvailable versions:")
for v in list_versions(s3, "invoice-001"):
    print(f"  {v['timestamp']} ({v['size']} bytes)")

latest = get_latest_version(s3, "invoice-001")
print(f"\nLatest version: run={latest['run']}, score={latest['score']}")

Exercise 3: Batch upload of documents

Write a script that uploads all the .txt files from a local directory to an S3 bucket, preserving the subdirectory structure. The script should report how many files it uploaded and the total size.

See solution 
import boto3
import os

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

s3.create_bucket(Bucket="batch-upload")


def upload_directory(s3, local_dir, bucket, s3_prefix="", extension=".txt"):
    """Uploads all files with the given extension from a directory to S3."""
    uploaded = 0
    total_bytes = 0

    for root, dirs, files in os.walk(local_dir):
        for filename in files:
            if not filename.endswith(extension):
                continue

            local_path = os.path.join(root, filename)
            relative_path = os.path.relpath(local_path, local_dir)
            s3_key = f"{s3_prefix}{relative_path}" if s3_prefix else relative_path

            file_size = os.path.getsize(local_path)
            s3.upload_file(local_path, bucket, s3_key)

            uploaded += 1
            total_bytes += file_size
            print(f"  Uploaded: {s3_key} ({file_size} bytes)")

    return uploaded, total_bytes


# Create test files
os.makedirs("data/test-batch/sub", exist_ok=True)
for i in range(5):
    path = f"data/test-batch/doc-{i}.txt"
    with open(path, "w") as f:
        f.write(f"Content of document {i} for AI analysis")
with open("data/test-batch/sub/nested.txt", "w") as f:
    f.write("Document in a subdirectory")

count, size = upload_directory(s3, "data/test-batch", "batch-upload", "documents/")
print(f"\nTotal: {count} files, {size} bytes")

# Verify
response = s3.list_objects_v2(Bucket="batch-upload")
for obj in response.get("Contents", []):
    print(f"  S3: {obj['Key']}")

Exercise 4: Presigned URL to share results

Create a system that generates presigned URLs for all the results in a bucket. The function should return a dictionary with the file name and its temporary URL (expires in 1 hour). Verify that the URLs work by downloading with requests.

See solution 
import boto3
import json
import requests

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

s3.create_bucket(Bucket="shared-results")

# Upload test results
for i in range(3):
    result = {"analysis": f"Result of analysis #{i+1}", "score": 0.9 + i * 0.02}
    s3.put_object(
        Bucket="shared-results",
        Key=f"results/analysis-{i+1}.json",
        Body=json.dumps(result),
        ContentType="application/json",
    )


def generate_share_links(s3, bucket, prefix="results/", expires_in=3600):
    """Generates presigned URLs for all objects with the given prefix."""
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    links = {}

    for obj in response.get("Contents", []):
        filename = obj["Key"].split("/")[-1]
        url = s3.generate_presigned_url(
            "get_object",
            Params={"Bucket": bucket, "Key": obj["Key"]},
            ExpiresIn=expires_in,
        )
        links[filename] = url

    return links


links = generate_share_links(s3, "shared-results")
print("Download links:")
for name, url in links.items():
    print(f"  {name}: {url[:80]}...")

# Verify they work
for name, url in links.items():
    resp = requests.get(url)
    data = resp.json()
    print(f"  Downloaded {name}: score={data['score']}")

Troubleshooting

"NoSuchBucket when calling put_object"

# The bucket doesn't exist — create it first
awslocal s3 mb s3://my-bucket

# Or in Python:
s3.create_bucket(Bucket="my-bucket")

"The files disappeared after restarting LocalStack"

# In the Community Edition, data doesn't persist by default
# Solution 1: use a Docker volume (limited)
volumes:
  - localstack_data:/var/lib/localstack

# Solution 2: use init scripts to recreate on startup
# (Recommended for development)

"Presigned URLs don't work from the browser"

# LocalStack generates URLs with localhost — they work from your machine
# But if your browser is on another host, you need your machine's IP
# Alternative: configure S3_HOSTNAME in LocalStack
environment:
  - HOSTNAME_EXTERNAL=192.168.1.100

"UnicodeEncodeError when uploading text with accents"

# Always encode explicitly as UTF-8
s3.put_object(
    Bucket="bucket",
    Key="file.txt",
    Body=text_with_accents.encode("utf-8"),
    ContentType="text/plain; charset=utf-8",
)

"list_objects_v2 only returns 1000 objects"

# S3 paginates the results at 1000 objects per page
# Use pagination:
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="bucket"):
    for obj in page.get("Contents", []):
        print(obj["Key"])

Summary

  • S3 on LocalStack works identically to real S3 — same APIs, same boto3 methods, same patterns.
  • Buckets organize your data: ai-input for documents, ai-output for results, ai-config for configuration.
  • put_object to upload, get_object to download, list_objects_v2 to list, generate_presigned_url to share.
  • The AI patterns include: Config Store (load prompts from S3), Result Store (save analyses), Batch Processing (process multiple documents).
  • Presigned URLs give temporary access without exposing credentials — useful for sharing analysis results.
  • All the code works the same against LocalStack and real AWS — capsule 06 teaches you the switch.

Additional Resources

  1. boto3 S3 Client Reference — Complete S3 API in boto3
  2. S3 Presigned URLs — Official presigned URLs documentation
  3. LocalStack S3 Coverage — Which S3 operations LocalStack supports
  4. S3 Best Practices — S3 performance optimization
  5. boto3 Paginators — Handling pagination in boto3
  6. S3 Storage Classes — Storage classes (for when you migrate to AWS)
← Previous2. LocalStack Setup and ConfigurationNext →4. Lambda Locally with LocalStack