Módulo 4: LocalStack — AWS Local Development

3. S3 Local para AI

Siguiente →

Descripción

En esta cápsula vas a dominar S3 en LocalStack con boto3: crear buckets, subir y descargar archivos, listar objetos, y generar presigned URLs. Cada ejemplo es AI-specific: almacenas embeddings, prompts, configuraciones de modelo, y resultados de inferencia. No es un tutorial genérico de S3 — es S3 como capa de almacenamiento para tu pipeline AI.

Contexto: En la cápsula anterior configuraste LocalStack en tu Docker Compose. Ahora tienes un S3 local funcionando. Aquí aprendes a operarlo con Python. Los patrones que implementas aquí se reutilizan en la cápsula 05 (pipeline S3 + Lambda) y en el proyecto final (cápsula 08). Todo lo que haces contra LocalStack funciona idéntico contra AWS real — solo cambia el endpoint URL.

S3 Fundamentals para AI

Qué es S3 (en contexto AI)

S3 (Simple Storage Service) es el servicio de almacenamiento de objetos de AWS. Para sistemas AI, S3 es donde almacenas:

  • Datos de entrada: Documentos a procesar, imágenes a analizar, audio a transcribir
  • Configuraciones: System prompts, parámetros de modelo, templates de respuesta
  • Resultados: Responses de LLMs, embeddings generados, análisis completados
  • Assets de modelo: Pesos de modelo (si usas modelos propios), vectores de embeddings
Pipeline AI típico con S3:
    ┌──────────────┐     ┌──────────────┐     ┌──────────────┐
    │  S3 Input    │ ──→ │  Procesador  │ ──→ │  S3 Output   │
    │              │     │  (Lambda +   │     │              │
    │ documentos/  │     │   LLM API)   │     │ resultados/  │
    │ prompts/     │     │              │     │ análisis/    │
    │ config/      │     │              │     │ embeddings/  │
    └──────────────┘     └──────────────┘     └──────────────┘

Conceptos clave

Bucket:
├── Contenedor principal de objetos (como un directorio raíz)
├── Nombre debe ser único (en AWS global, en LocalStack local)
├── Ejemplo: "ai-input", "ai-output", "ai-models"

Object (Objeto):
├── Un archivo almacenado en un bucket
├── Identificado por Key (ruta dentro del bucket)
├── Ejemplo: Key = "prompts/system-v2.txt"
├── No hay carpetas reales — el "/" en la key es convención

Key:
├── La "ruta" del objeto dentro del bucket
├── Ejemplo: "documents/factura-001.pdf"
├── Prefijos simulan carpetas: "documents/", "results/"

Conectar boto3 a LocalStack

Client de S3

import boto3
import json

def get_s3_client(endpoint_url="http://localhost:4566"):
    """Crea un cliente S3 para 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()

Este patrón — una función que acepta endpoint_url como parámetro — es la base del environment switching que verás en la cápsula 06. Cambias la URL y tu código apunta a AWS real.

Resource vs Client

boto3 tiene dos interfaces: client (bajo nivel) y resource (alto nivel). Para operaciones AI, client es más explícito y predecible:

# Client (recomendado — explícito, predecible)
s3_client = boto3.client("s3", endpoint_url="http://localhost:4566")
s3_client.put_object(Bucket="my-bucket", Key="file.json", Body=data)

# Resource (alto nivel — más "pythonic" pero menos control)
s3_resource = boto3.resource("s3", endpoint_url="http://localhost:4566")
s3_resource.Bucket("my-bucket").put_object(Key="file.json", Body=data)

En esta guía usamos client porque es lo que verás en código de producción y en la documentación de AWS.

Operaciones con Buckets

Crear buckets

s3 = get_s3_client()

# Crear buckets para tu pipeline AI
buckets = ["ai-input", "ai-output", "ai-config"]

for bucket_name in buckets:
    try:
        s3.create_bucket(Bucket=bucket_name)
        print(f"Bucket '{bucket_name}' creado")
    except s3.exceptions.BucketAlreadyOwnedByYou:
        print(f"Bucket '{bucket_name}' ya existe")

Listar buckets

response = s3.list_buckets()

print("Buckets disponibles:")
for bucket in response["Buckets"]:
    print(f"  - {bucket['Name']} (creado: {bucket['CreationDate']})")

Eliminar bucket

# Un bucket debe estar vacío para eliminarlo
# Primero borra todos los objetos
def empty_and_delete_bucket(s3, bucket_name):
    """Vacía y elimina un 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"  Eliminado: {obj['Key']}")

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

Subir Archivos (Upload)

Subir texto (prompts, configs)

# Subir un system prompt
system_prompt = """Eres un asistente especializado en análisis de documentos.
Tu tarea es extraer información clave de documentos de negocio.
Responde siempre en formato JSON con los campos: titulo, resumen, puntos_clave."""

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

Subir JSON (configuraciones de modelo, resultados)

# Subir configuración de modelo
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("Configuración de modelo subida")

Subir archivos binarios (documentos a procesar)

# Subir un documento desde disco
with open("data/factura-ejemplo.pdf", "rb") as f:
    s3.put_object(
        Bucket="ai-input",
        Key="documents/factura-001.pdf",
        Body=f.read(),
        ContentType="application/pdf",
    )
print("Documento subido para procesamiento")

# Alternativa: upload_file (más eficiente para archivos grandes)
s3.upload_file(
    Filename="data/factura-ejemplo.pdf",
    Bucket="ai-input",
    Key="documents/factura-002.pdf",
)

Subir resultados de inferencia

# Resultado de un análisis AI
analysis_result = {
    "document_key": "documents/factura-001.pdf",
    "model": "gpt-4o-mini",
    "analysis": {
        "titulo": "Factura #12345",
        "resumen": "Factura de servicios de consultoría por $5,000 USD",
        "puntos_clave": [
            "Proveedor: Acme Corp",
            "Fecha: 2026-03-01",
            "Total: $5,000 USD",
            "Concepto: Consultoría técnica",
        ],
    },
    "tokens_used": 342,
    "duration_ms": 2100,
    "processed_at": "2026-03-08T15:30:00Z",
}

s3.put_object(
    Bucket="ai-output",
    Key="results/factura-001-analysis.json",
    Body=json.dumps(analysis_result, indent=2, ensure_ascii=False),
    ContentType="application/json",
)
print("Resultado de análisis guardado en S3")

Descargar Archivos (Download)

Descargar como string

# Descargar y leer un archivo de texto
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}")

Descargar JSON

# Descargar y parsear JSON
response = s3.get_object(
    Bucket="ai-output",
    Key="results/factura-001-analysis.json",
)
result = json.loads(response["Body"].read().decode("utf-8"))
print(f"Título: {result['analysis']['titulo']}")
print(f"Resumen: {result['analysis']['resumen']}")

Descargar a archivo local

# Descargar a disco
s3.download_file(
    Bucket="ai-output",
    Key="results/factura-001-analysis.json",
    Filename="data/output/factura-001-analysis.json",
)
print("Archivo descargado a data/output/")

Verificar si un objeto existe

def object_exists(s3, bucket, key):
    """Verifica si un objeto existe en S3."""
    try:
        s3.head_object(Bucket=bucket, Key=key)
        return True
    except s3.exceptions.ClientError:
        return False

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

Listar Objetos

Listar todos los objetos de un bucket

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

if "Contents" in response:
    print(f"Objetos en ai-input ({response['KeyCount']}):")
    for obj in response["Contents"]:
        size_kb = obj["Size"] / 1024
        print(f"  {obj['Key']} ({size_kb:.1f} KB)")
else:
    print("Bucket vacío")

Listar por prefijo (simular carpetas)

# Solo objetos en "documents/"
response = s3.list_objects_v2(
    Bucket="ai-input",
    Prefix="documents/",
)

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

# Solo objetos en "results/"
response = s3.list_objects_v2(
    Bucket="ai-output",
    Prefix="results/",
)

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

Paginación para buckets con muchos objetos

def list_all_objects(s3, bucket, prefix=""):
    """Lista todos los objetos, manejando paginación."""
    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 de resultados: {len(objects)}")

Presigned URLs

Qué son y para qué sirven

Presigned URLs permiten dar acceso temporal a un objeto S3 sin exponer credenciales. Son útiles cuando tu app AI necesita:

  • Permitir que un usuario descargue un resultado sin autenticarse en AWS
  • Generar un link temporal para compartir un análisis
  • Permitir upload directo desde un browser

Generar URL de descarga

# Generar URL temporal para descargar un resultado
url = s3.generate_presigned_url(
    "get_object",
    Params={
        "Bucket": "ai-output",
        "Key": "results/factura-001-analysis.json",
    },
    ExpiresIn=3600,  # 1 hora
)

print(f"URL de descarga (expira en 1h):\n{url}")
# En LocalStack: http://localhost:4566/ai-output/results/...?X-Amz-...

Generar URL de upload

# Generar URL temporal para subir un archivo
upload_url = s3.generate_presigned_url(
    "put_object",
    Params={
        "Bucket": "ai-input",
        "Key": "documents/nuevo-documento.pdf",
        "ContentType": "application/pdf",
    },
    ExpiresIn=900,  # 15 minutos
)

print(f"URL de upload (expira en 15min):\n{upload_url}")

Usar presigned URL con requests

import requests

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

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

Patrones AI con S3

Patrón 1: Config Store — cargar configuración de modelo desde S3

def load_model_config(s3, bucket="ai-config", key="models/analyzer-config.json"):
    """Carga configuración de modelo desde 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"Modelo: {config['model']}, Temp: {config['temperature']}")

Patrón 2: Result Store — guardar y recuperar resultados de inferencia

import datetime

def save_inference_result(s3, document_key, result, bucket="ai-output"):
    """Guarda el resultado de inferencia en 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/factura-001.pdf", analysis_result)
print(f"Resultado guardado en: {output_key}")

Patrón 3: Batch Processing — procesar múltiples documentos

def get_pending_documents(s3, bucket="ai-input", prefix="documents/"):
    """Lista documentos pendientes de procesar."""
    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"):
    """Verifica si un documento ya fue procesado."""
    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"Pendiente: {doc_key}")
    else:
        print(f"Ya procesado: {doc_key}")

Ejercicios

Ejercicio 1: Crear un repositorio de prompts en S3

Crea un bucket prompt-library y sube 3 system prompts diferentes (analyzer, summarizer, translator). Luego escribe una función que reciba el nombre del prompt y lo descargue de S3.

Ver solución 
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": "Analiza el siguiente documento y extrae: título, resumen, puntos clave. Responde en JSON.",
    "summarizer": "Resume el siguiente texto en máximo 3 oraciones. Sé conciso y preciso.",
    "translator": "Traduce el siguiente texto al inglés. Mantén el tono y formato original.",
}

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}' subido")


def get_prompt(s3, prompt_name, bucket="prompt-library"):
    """Descarga un system prompt por nombre."""
    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}' no encontrado")


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

Ejercicio 2: Versionado de resultados

Implementa un sistema que guarde múltiples versiones de un resultado de análisis para el mismo documento. Cada ejecución del análisis crea un nuevo archivo con timestamp. Escribe una función que liste todas las versiones y otra que descargue la más reciente.

Ver solución 
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"):
    """Guarda resultado con timestamp como versión."""
    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"):
    """Lista todas las versiones de un resultado."""
    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"):
    """Descarga la versión más reciente."""
    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"))


# Simular 3 ejecuciones
import time

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

print("\nVersiones disponibles:")
for v in list_versions(s3, "factura-001"):
    print(f"  {v['timestamp']} ({v['size']} bytes)")

latest = get_latest_version(s3, "factura-001")
print(f"\nÚltima versión: run={latest['run']}, score={latest['score']}")

Ejercicio 3: Upload de batch de documentos

Escribe un script que suba todos los archivos .txt de un directorio local a un bucket S3, preservando la estructura de subdirectorios. El script debe reportar cuántos archivos subió y el tamaño total.

Ver solución 
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"):
    """Sube todos los archivos con extensión dada de un directorio a 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"  Subido: {s3_key} ({file_size} bytes)")

    return uploaded, total_bytes


# Crear archivos de prueba
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"Contenido del documento {i} para análisis AI")
with open("data/test-batch/sub/nested.txt", "w") as f:
    f.write("Documento en subdirectorio")

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

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

Ejercicio 4: Presigned URL para compartir resultados

Crea un sistema que genere presigned URLs para todos los resultados en un bucket. La función debe retornar un diccionario con el nombre del archivo y su URL temporal (expira en 1 hora). Verifica que las URLs funcionan descargando con requests.

Ver solución 
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")

# Subir resultados de prueba
for i in range(3):
    result = {"analysis": f"Resultado del análisis #{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):
    """Genera presigned URLs para todos los objetos con el prefijo dado."""
    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("Links de descarga:")
for name, url in links.items():
    print(f"  {name}: {url[:80]}...")

# Verificar que funcionan
for name, url in links.items():
    resp = requests.get(url)
    data = resp.json()
    print(f"  Descargado {name}: score={data['score']}")

Troubleshooting

"NoSuchBucket al hacer put_object"

# El bucket no existe — créalo primero
awslocal s3 mb s3://mi-bucket

# O en Python:
s3.create_bucket(Bucket="mi-bucket")

"Los archivos desaparecieron después de reiniciar LocalStack"

# En Community Edition, los datos no persisten por default
# Solución 1: usa volumen Docker (limitado)
volumes:
  - localstack_data:/var/lib/localstack

# Solución 2: usa init scripts para recrear al arrancar
# (Recomendado para desarrollo)

"Presigned URLs no funcionan desde el browser"

# LocalStack genera URLs con localhost — funcionan desde tu máquina
# Pero si tu browser está en otro host, necesitas la IP de tu máquina
# Alternativa: configura S3_HOSTNAME en LocalStack
environment:
  - HOSTNAME_EXTERNAL=192.168.1.100

"UnicodeEncodeError al subir texto con acentos"

# Siempre codifica explícitamente como UTF-8
s3.put_object(
    Bucket="bucket",
    Key="archivo.txt",
    Body=texto_con_acentos.encode("utf-8"),
    ContentType="text/plain; charset=utf-8",
)

"list_objects_v2 solo retorna 1000 objetos"

# S3 pagina los resultados a 1000 objetos por página
# Usa paginación:
paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket="bucket"):
    for obj in page.get("Contents", []):
        print(obj["Key"])

Resumen

  • S3 en LocalStack funciona idéntico a S3 real — mismas APIs, mismos métodos boto3, mismos patrones.
  • Buckets organizan tus datos: ai-input para documentos, ai-output para resultados, ai-config para configuración.
  • put_object para subir, get_object para descargar, list_objects_v2 para listar, generate_presigned_url para compartir.
  • Los patrones AI incluyen: Config Store (cargar prompts de S3), Result Store (guardar análisis), Batch Processing (procesar múltiples documentos).
  • Presigned URLs dan acceso temporal sin exponer credenciales — útil para compartir resultados de análisis.
  • Todo el código funciona igual contra LocalStack y AWS real — la cápsula 06 te enseña el switch.

Recursos Adicionales

  1. boto3 S3 Client Reference — API completa de S3 en boto3
  2. S3 Presigned URLs — Documentación oficial de presigned URLs
  3. LocalStack S3 Coverage — Qué operaciones S3 soporta LocalStack
  4. S3 Best Practices — Optimización de rendimiento S3
  5. boto3 Paginators — Manejo de paginación en boto3
  6. S3 Storage Classes — Clases de almacenamiento (para cuando migres a AWS)
← Anterior2. LocalStack Setup y ConfiguraciónSiguiente →4. Lambda Local con LocalStack