2. S3 para AI Assets

Descripción

En esta cápsula vas a dominar S3 como data layer de tu sistema AI. S3 no es "almacenamiento genérico" — es donde viven los assets que hacen funcionar tu sistema: modelos serializados, archivos de embeddings pre-computados, documentos para RAG, prompt templates versionados, y logs de inferencia. Aprenderás a organizar un bucket para proyectos AI, a ejecutar operaciones con boto3 (put_object, get_object, list_objects_v2, presigned URLs), y a diseñar patrones de almacenamiento que escalen.

Contexto: En el Módulo 4 creaste buckets en LocalStack y subiste archivos de prueba. Ahora el foco cambia de "cómo usar S3" a "cómo usar S3 para AI." Cada operación que aprendas aquí tiene un propósito AI-specific: no subes un .txt cualquiera, subes un prompt template que tu Lambda leerá para construir inferencia. Al terminar, tendrás un módulo Python reutilizable para gestionar todos los assets de AI de tu sistema.

S3 Concepts para AI Engineers

Buckets, objetos, y prefijos

S3 Bucket: ai-assets-123456789012
├── prompts/                          ← Prefijo (no es una "carpeta" real)
│   ├── summarizer/v1/system.txt      ← Objeto (prompt template)
│   ├── summarizer/v2/system.txt
│   └── classifier/v1/system.txt
├── documents/                        ← RAG knowledge base
│   ├── product-docs/manual-v3.pdf
│   └── faq/faq-2026-q1.json
├── embeddings/                       ← Vectores pre-computados
│   ├── product-docs/embeddings.npy
│   └── faq/embeddings.npy
├── models/                           ← Modelos serializados
│   └── classifier/sentiment-v2.pkl
└── responses/                        ← Logs de inferencia
    └── 2026/03/08/
        ├── resp-001.json
        └── resp-002.json

Conceptos clave:

• Bucket: Contenedor de nivel superior. Nombre globalmente único en AWS. Uno por proyecto o por entorno (dev, staging, prod).
• Objeto: Un archivo almacenado en S3. Cada objeto tiene una key (ruta), contenido (bytes), y metadata.
• Prefijo: S3 no tiene carpetas reales. prompts/summarizer/v1/system.txt es una key plana. Los "/" son convención para organizar.
• Versioning: S3 puede mantener múltiples versiones del mismo objeto. Útil para prompt templates: rollback a versión anterior si la nueva empeora resultados.

¿Por qué S3 para AI y no una base de datos?

Regla práctica: Archivos + acceso por key + triggers → S3. Datos estructurados + queries complejas → Base de datos. Para AI, la mayoría de assets son archivos (modelos, embeddings, documentos), así que S3 es la opción natural.

Organización de Bucket para Proyectos AI

Patrón recomendado

BUCKET_STRUCTURE = {
    "prompts/": "Prompt templates versionados. Tu Lambda los lee antes de invocar al LLM.",
    "documents/": "Documentos fuente para RAG. PDFs, JSONs, markdown.",
    "embeddings/": "Vectores pre-computados. Se generan offline, se consumen en inferencia.",
    "models/": "Modelos serializados (pickle, joblib, ONNX). Para inferencia local o SageMaker.",
    "responses/": "Logs de inferencia. Cada respuesta del LLM se guarda para auditoría.",
    "config/": "Configuración del sistema. Feature flags, parámetros de modelo.",
}

Naming conventions

# ✅ Buenos nombres de key
"prompts/summarizer/v2/system.txt"           # Versionado explícito
"documents/product-docs/manual-2026-q1.pdf"  # Temporal
"responses/2026/03/08/req-a1b2c3.json"       # Particionado por fecha
"embeddings/product-docs/chunk-embeddings.npy"

# ❌ Malos nombres de key
"prompt.txt"                    # Sin contexto
"data/file1.json"               # No descriptivo
"embeddings.npy"                # Sin organización
"responses/latest.json"         # Se sobrescribe

Bucket por entorno

# Opción 1: Un bucket por entorno (recomendado para equipos)
BUCKETS = {
    "dev": "ai-assets-dev-123456789012",
    "staging": "ai-assets-staging-123456789012",
    "prod": "ai-assets-prod-123456789012",
}

# Opción 2: Un bucket con prefijo por entorno (más simple)
BUCKET = "ai-assets-123456789012"
PREFIXES = {
    "dev": "dev/",
    "staging": "staging/",
    "prod": "prod/",
}

Operaciones S3 con boto3

Setup del cliente

import boto3
import json
import os
from datetime import datetime

def get_s3_client():
    """Crea un cliente S3 compatible con LocalStack y AWS."""
    endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
    if endpoint_url:
        return boto3.client(
            "s3",
            endpoint_url=endpoint_url,
            aws_access_key_id=os.environ.get("AWS_ACCESS_KEY_ID", "test"),
            aws_secret_access_key=os.environ.get("AWS_SECRET_ACCESS_KEY", "test"),
            region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
        )
    return boto3.client("s3")

s3 = get_s3_client()

Crear bucket

def create_ai_bucket(bucket_name: str, region: str = "us-east-1") -> dict:
    """Crea un bucket S3 para assets de AI."""
    create_params = {"Bucket": bucket_name}

    if region != "us-east-1":
        create_params["CreateBucketConfiguration"] = {
            "LocationConstraint": region
        }

    response = s3.create_bucket(**create_params)
    print(f"Bucket creado: {bucket_name}")
    return response


bucket_name = f"ai-assets-{os.environ.get('AWS_ACCOUNT_ID', 'dev')}"
create_ai_bucket(bucket_name)

put_object — Subir assets de AI

def upload_prompt_template(
    bucket: str, name: str, version: str, content: str
) -> dict:
    """Sube un prompt template versionado a S3."""
    key = f"prompts/{name}/{version}/system.txt"
    response = s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=content.encode("utf-8"),
        ContentType="text/plain",
        Metadata={
            "template-name": name,
            "version": version,
            "created-at": datetime.utcnow().isoformat(),
        },
    )
    print(f"Prompt template subido: s3://{bucket}/{key}")
    return response


upload_prompt_template(
    bucket=bucket_name,
    name="summarizer",
    version="v1",
    content=(
        "Eres un asistente especializado en generar resúmenes.\n"
        "Genera un resumen conciso del siguiente documento.\n"
        "Máximo 3 párrafos. Usa lenguaje claro y directo.\n"
        "Incluye los puntos más importantes."
    ),
)

def upload_rag_document(bucket: str, collection: str, doc: dict) -> dict:
    """Sube un documento para RAG a S3."""
    doc_id = doc.get("id", datetime.utcnow().strftime("%Y%m%d%H%M%S"))
    key = f"documents/{collection}/{doc_id}.json"

    response = s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=json.dumps(doc, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
        Metadata={
            "collection": collection,
            "doc-id": doc_id,
        },
    )
    print(f"Documento RAG subido: s3://{bucket}/{key}")
    return response


upload_rag_document(
    bucket=bucket_name,
    collection="product-docs",
    doc={
        "id": "doc-001",
        "title": "Guía de instalación",
        "content": "Para instalar el sistema, sigue estos pasos...",
        "metadata": {"category": "setup", "language": "es"},
    },
)

def upload_inference_response(
    bucket: str, request_id: str, response_data: dict
) -> dict:
    """Guarda la respuesta de una inferencia para auditoría."""
    now = datetime.utcnow()
    key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"

    payload = {
        "request_id": request_id,
        "timestamp": now.isoformat(),
        **response_data,
    }

    result = s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        ContentType="application/json",
    )
    print(f"Response guardada: s3://{bucket}/{key}")
    return result

get_object — Recuperar assets

def get_prompt_template(bucket: str, name: str, version: str) -> str:
    """Lee un prompt template de S3."""
    key = f"prompts/{name}/{version}/system.txt"
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response["Body"].read().decode("utf-8")
    return content


template = get_prompt_template(bucket_name, "summarizer", "v1")
print(template)

def get_rag_document(bucket: str, collection: str, doc_id: str) -> dict:
    """Lee un documento RAG de S3."""
    key = f"documents/{collection}/{doc_id}.json"
    response = s3.get_object(Bucket=bucket, Key=key)
    content = response["Body"].read().decode("utf-8")
    return json.loads(content)


doc = get_rag_document(bucket_name, "product-docs", "doc-001")
print(doc["title"])

list_objects_v2 — Listar assets

def list_ai_assets(bucket: str, prefix: str, max_keys: int = 100) -> list:
    """Lista assets de AI por prefijo."""
    response = s3.list_objects_v2(
        Bucket=bucket,
        Prefix=prefix,
        MaxKeys=max_keys,
    )

    objects = []
    for obj in response.get("Contents", []):
        objects.append({
            "key": obj["Key"],
            "size_bytes": obj["Size"],
            "last_modified": obj["LastModified"].isoformat(),
        })

    return objects


prompts = list_ai_assets(bucket_name, "prompts/")
for p in prompts:
    print(f"  {p['key']} ({p['size_bytes']} bytes)")

def list_all_with_pagination(bucket: str, prefix: str) -> list:
    """Lista todos los objetos, manejando paginación para >1000 resultados."""
    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["Key"])

    return all_objects

generate_presigned_url — URLs temporales

def generate_download_url(
    bucket: str, key: str, expires_in: int = 3600
) -> str:
    """Genera una URL temporal para descargar un objeto sin credenciales AWS."""
    url = s3.generate_presigned_url(
        "get_object",
        Params={"Bucket": bucket, "Key": key},
        ExpiresIn=expires_in,
    )
    return url


url = generate_download_url(bucket_name, "documents/product-docs/doc-001.json")
print(f"URL temporal (1h): {url}")

Presigned URLs son útiles cuando:

• Tu frontend necesita descargar un documento RAG directamente de S3 (sin pasar por Lambda)
• Quieres compartir un modelo serializado con un compañero de equipo
• Una Lambda genera un reporte y envía el link por email/webhook

def generate_upload_url(
    bucket: str, key: str, content_type: str = "application/json", expires_in: int = 900
) -> str:
    """Genera una URL temporal para que un cliente suba un archivo directo a S3."""
    url = s3.generate_presigned_url(
        "put_object",
        Params={
            "Bucket": bucket,
            "Key": key,
            "ContentType": content_type,
        },
        ExpiresIn=expires_in,
    )
    return url


upload_url = generate_upload_url(
    bucket_name,
    "documents/uploads/user-doc-abc123.pdf",
    content_type="application/pdf",
)
print(f"Upload URL (15min): {upload_url}")

delete_object — Limpiar assets obsoletos

def delete_old_responses(bucket: str, prefix: str, dry_run: bool = True) -> list:
    """Elimina responses antiguas de S3."""
    objects = list_all_with_pagination(bucket, prefix)

    if dry_run:
        print(f"[DRY RUN] Se eliminarían {len(objects)} objetos:")
        for key in objects[:5]:
            print(f"  - {key}")
        if len(objects) > 5:
            print(f"  ... y {len(objects) - 5} más")
        return objects

    if not objects:
        return []

    delete_keys = [{"Key": key} for key in objects]
    response = s3.delete_objects(
        Bucket=bucket,
        Delete={"Objects": delete_keys},
    )
    deleted = response.get("Deleted", [])
    print(f"Eliminados: {len(deleted)} objetos")
    return [d["Key"] for d in deleted]

Módulo Reutilizable: S3 AI Assets Manager

"""s3_ai_assets.py — Gestión de AI assets en S3."""

import boto3
import json
import os
from datetime import datetime
from typing import Optional


class S3AIAssets:
    """Gestiona assets de AI en un bucket S3."""

    def __init__(self, bucket_name: str, endpoint_url: Optional[str] = None):
        self.bucket = bucket_name
        client_kwargs = {}
        if endpoint_url:
            client_kwargs["endpoint_url"] = endpoint_url
        self.s3 = boto3.client("s3", **client_kwargs)

    def put_prompt(self, name: str, version: str, content: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=content.encode("utf-8"),
            ContentType="text/plain",
            Metadata={"version": version, "updated": datetime.utcnow().isoformat()},
        )
        return key

    def get_prompt(self, name: str, version: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        resp = self.s3.get_object(Bucket=self.bucket, Key=key)
        return resp["Body"].read().decode("utf-8")

    def put_document(self, collection: str, doc_id: str, data: dict) -> str:
        key = f"documents/{collection}/{doc_id}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(data, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def get_document(self, collection: str, doc_id: str) -> dict:
        key = f"documents/{collection}/{doc_id}.json"
        resp = self.s3.get_object(Bucket=self.bucket, Key=key)
        return json.loads(resp["Body"].read().decode("utf-8"))

    def put_response(self, request_id: str, data: dict) -> str:
        now = datetime.utcnow()
        key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
        payload = {"request_id": request_id, "timestamp": now.isoformat(), **data}
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def list_keys(self, prefix: str) -> list:
        paginator = self.s3.get_paginator("list_objects_v2")
        keys = []
        for page in paginator.paginate(Bucket=self.bucket, Prefix=prefix):
            for obj in page.get("Contents", []):
                keys.append(obj["Key"])
        return keys

    def presigned_download(self, key: str, expires: int = 3600) -> str:
        return self.s3.generate_presigned_url(
            "get_object",
            Params={"Bucket": self.bucket, "Key": key},
            ExpiresIn=expires,
        )


# Uso
assets = S3AIAssets(
    bucket_name="ai-assets-dev",
    endpoint_url=os.environ.get("AWS_ENDPOINT_URL"),
)

assets.put_prompt("summarizer", "v1", "Resume el siguiente documento en 3 puntos.")
template = assets.get_prompt("summarizer", "v1")
print(template)

S3 Lifecycle: Gestión Automática de Costes

Para un sistema AI en producción, los responses se acumulan. Lifecycle rules automatizan la transición a storage más barato o la eliminación:

def configure_ai_lifecycle(bucket: str) -> dict:
    """Configura lifecycle rules para assets de AI."""
    response = s3.put_bucket_lifecycle_configuration(
        Bucket=bucket,
        LifecycleConfiguration={
            "Rules": [
                {
                    "ID": "archive-old-responses",
                    "Filter": {"Prefix": "responses/"},
                    "Status": "Enabled",
                    "Transitions": [
                        {
                            "Days": 30,
                            "StorageClass": "STANDARD_IA",
                        },
                        {
                            "Days": 90,
                            "StorageClass": "GLACIER",
                        },
                    ],
                    "Expiration": {"Days": 365},
                },
                {
                    "ID": "clean-temp-uploads",
                    "Filter": {"Prefix": "temp/"},
                    "Status": "Enabled",
                    "Expiration": {"Days": 7},
                },
            ],
        },
    )
    print(f"Lifecycle configurado para {bucket}")
    return response

Resultado del lifecycle:
responses/*
├── 0-30 días:   S3 Standard     ($0.023/GB)
├── 30-90 días:  S3 Standard-IA  ($0.0125/GB)  ← 46% más barato
├── 90-365 días: S3 Glacier      ($0.004/GB)   ← 83% más barato
└── >365 días:   Eliminado automáticamente

temp/*
└── >7 días:     Eliminado automáticamente

Troubleshooting

Problema 1: "NoSuchBucket" al hacer put_object

El bucket no existe. Necesitas crearlo antes de subir objetos.

try:
    s3.head_bucket(Bucket=bucket_name)
except s3.exceptions.ClientError as e:
    if e.response["Error"]["Code"] == "404":
        print(f"Bucket {bucket_name} no existe. Creándolo...")
        s3.create_bucket(Bucket=bucket_name)
    else:
        raise

Problema 2: "AccessDenied" al leer de S3

Tu rol IAM no tiene permisos para la operación. En LocalStack no pasa; en AWS real, sí.

# Verificar permisos del rol
aws iam get-role-policy --role-name tu-lambda-role --policy-name s3-access

# El policy necesita s3:GetObject en el bucket específico

Problema 3: get_object retorna bytes, no string

response["Body"].read() retorna bytes. Necesitas decodificar.

# ❌ Error común
content = s3.get_object(Bucket=b, Key=k)["Body"].read()
data = json.loads(content)  # Funciona porque json.loads acepta bytes

# ✅ Explícito y seguro
content = s3.get_object(Bucket=b, Key=k)["Body"].read().decode("utf-8")
data = json.loads(content)

Problema 4: Presigned URLs no funcionan en LocalStack

LocalStack genera URLs con localhost, que no son accesibles desde fuera de tu máquina.

# En LocalStack, la URL generada es:
# http://localhost:4566/bucket/key?...

# Si accedes desde un container Docker, usa el nombre del servicio:
# http://localstack:4566/bucket/key?...

# Para testing local, localhost funciona.

Problema 5: list_objects_v2 retorna solo 1000 objetos

S3 pagina a 1000 objetos por request. Usa el paginator de boto3.

paginator = s3.get_paginator("list_objects_v2")
for page in paginator.paginate(Bucket=bucket, Prefix="responses/"):
    for obj in page.get("Contents", []):
        print(obj["Key"])

Ejercicios Prácticos

Ejercicio 1: Versionado de prompt templates

Crea una función que suba múltiples versiones de un prompt template a S3 y otra que recupere la última versión disponible (la de número más alto).

import boto3
import json
import os
import re

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = "ai-assets-dev"


def upload_prompt_versions(name: str, versions: dict[str, str]) -> list:
    """Sube múltiples versiones de un prompt. versions = {"v1": "...", "v2": "..."}"""
    keys = []
    for version, content in versions.items():
        key = f"prompts/{name}/{version}/system.txt"
        s3.put_object(
            Bucket=BUCKET,
            Key=key,
            Body=content.encode("utf-8"),
            ContentType="text/plain",
        )
        keys.append(key)
        print(f"Subido: {key}")
    return keys


def get_latest_prompt(name: str) -> tuple[str, str]:
    """Retorna (version, content) de la última versión del prompt."""
    response = s3.list_objects_v2(
        Bucket=BUCKET,
        Prefix=f"prompts/{name}/",
    )

    versions = []
    for obj in response.get("Contents", []):
        match = re.search(r"/v(\d+)/", obj["Key"])
        if match:
            versions.append((int(match.group(1)), obj["Key"]))

    if not versions:
        raise ValueError(f"No versions found for prompt '{name}'")

    versions.sort(key=lambda x: x[0], reverse=True)
    latest_key = versions[0][1]
    latest_version = f"v{versions[0][0]}"

    resp = s3.get_object(Bucket=BUCKET, Key=latest_key)
    content = resp["Body"].read().decode("utf-8")

    return latest_version, content


upload_prompt_versions("summarizer", {
    "v1": "Resume el documento en un párrafo.",
    "v2": "Resume el documento en 3 bullets. Sé conciso.",
    "v3": "Resume el documento en 3 bullets. Incluye datos numéricos si existen.",
})

version, content = get_latest_prompt("summarizer")
print(f"Última versión: {version}")
print(f"Contenido: {content}")

Ejercicio 2: Bulk upload de documentos RAG

Implementa una función que tome una lista de documentos (dicts) y los suba a S3 en una colección específica. Incluye un contador de éxitos y fallos.

import boto3
import json
import os
from dataclasses import dataclass

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = "ai-assets-dev"


@dataclass
class UploadResult:
    total: int
    success: int
    failed: int
    errors: list


def bulk_upload_documents(collection: str, documents: list[dict]) -> UploadResult:
    """Sube una lista de documentos a S3 para RAG."""
    result = UploadResult(total=len(documents), success=0, failed=0, errors=[])

    for doc in documents:
        doc_id = doc.get("id")
        if not doc_id:
            result.failed += 1
            result.errors.append({"doc": doc, "error": "Missing 'id' field"})
            continue

        key = f"documents/{collection}/{doc_id}.json"
        try:
            s3.put_object(
                Bucket=BUCKET,
                Key=key,
                Body=json.dumps(doc, ensure_ascii=False).encode("utf-8"),
                ContentType="application/json",
            )
            result.success += 1
        except Exception as e:
            result.failed += 1
            result.errors.append({"doc_id": doc_id, "error": str(e)})

    print(f"Upload completado: {result.success}/{result.total} exitosos")
    if result.errors:
        print(f"  Errores: {result.failed}")
    return result


docs = [
    {"id": "faq-001", "question": "¿Cómo instalo el SDK?", "answer": "Usa pip install..."},
    {"id": "faq-002", "question": "¿Cuál es el pricing?", "answer": "Plan free + plan pro..."},
    {"id": "faq-003", "question": "¿Hay API pública?", "answer": "Sí, documentada en..."},
]

result = bulk_upload_documents("faq", docs)
print(f"Resultado: {result}")

Ejercicio 3: Exportar colección RAG completa

Implementa una función que descargue todos los documentos de una colección RAG de S3 y los retorne como una lista de dicts. Debe manejar paginación.

import boto3
import json
import os

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = "ai-assets-dev"


def export_collection(collection: str) -> list[dict]:
    """Descarga todos los documentos de una colección RAG."""
    prefix = f"documents/{collection}/"
    paginator = s3.get_paginator("list_objects_v2")

    documents = []
    for page in paginator.paginate(Bucket=BUCKET, Prefix=prefix):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if not key.endswith(".json"):
                continue

            try:
                response = s3.get_object(Bucket=BUCKET, Key=key)
                content = response["Body"].read().decode("utf-8")
                doc = json.loads(content)
                doc["_s3_key"] = key
                doc["_s3_size"] = obj["Size"]
                doc["_s3_modified"] = obj["LastModified"].isoformat()
                documents.append(doc)
            except Exception as e:
                print(f"Error leyendo {key}: {e}")

    print(f"Exportados {len(documents)} documentos de '{collection}'")
    return documents


docs = export_collection("faq")
for doc in docs:
    print(f"  {doc['id']}: {doc.get('question', 'N/A')}")

Ejercicio 4: Auditoría de uso de storage

Crea una función que analice el contenido de un bucket y genere un reporte de uso por prefijo (tamaño total, número de objetos, objeto más reciente).

import boto3
import os
from collections import defaultdict
from datetime import datetime

s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))
BUCKET = "ai-assets-dev"


def storage_audit(bucket: str) -> dict:
    """Genera un reporte de uso de storage por prefijo de primer nivel."""
    paginator = s3.get_paginator("list_objects_v2")
    stats = defaultdict(lambda: {
        "count": 0,
        "total_bytes": 0,
        "latest_modified": None,
        "latest_key": None,
    })

    for page in paginator.paginate(Bucket=bucket):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            prefix = key.split("/")[0] + "/" if "/" in key else "(root)"

            stats[prefix]["count"] += 1
            stats[prefix]["total_bytes"] += obj["Size"]

            mod_time = obj["LastModified"]
            if (
                stats[prefix]["latest_modified"] is None
                or mod_time > stats[prefix]["latest_modified"]
            ):
                stats[prefix]["latest_modified"] = mod_time
                stats[prefix]["latest_key"] = key

    report = dict(stats)
    total_bytes = sum(s["total_bytes"] for s in report.values())
    total_objects = sum(s["count"] for s in report.values())

    print(f"=== Storage Audit: {bucket} ===")
    print(f"Total: {total_objects} objetos, {total_bytes / 1024:.1f} KB")
    print()
    for prefix, data in sorted(report.items()):
        size_kb = data["total_bytes"] / 1024
        print(f"  {prefix:<20} {data['count']:>5} objetos  {size_kb:>8.1f} KB")
        if data["latest_key"]:
            print(f"  {'':20} último: {data['latest_key']}")
    print()

    estimated_monthly = (total_bytes / (1024**3)) * 0.023
    print(f"Coste S3 estimado: ${estimated_monthly:.4f}/mes (S3 Standard)")

    return report


storage_audit(BUCKET)

Resumen

• S3 es el data layer de tu sistema AI. No es almacenamiento genérico — es donde viven tus prompt templates, documentos RAG, embeddings, modelos, y logs de inferencia.
• Organiza por prefijo con convención clara: prompts/, documents/, embeddings/, models/, responses/. Versiona explícitamente.
• boto3 operations clave: put_object para subir, get_object para leer, list_objects_v2 con paginator para listar, generate_presigned_url para URLs temporales.
• Lifecycle rules automatizan costes: Mueve responses antiguos a storage barato (IA, Glacier) y elimina archivos temporales automáticamente.
• El módulo S3AIAssets es reutilizable. Lo usarás en la cápsula 04 (integración S3+Lambda) y en el proyecto del módulo.
• Todo funciona igual en LocalStack y AWS. La diferencia es endpoint_url en boto3.

Recursos Adicionales

1. S3 User Guide — Documentación oficial completa
2. boto3 S3 Examples — Ejemplos oficiales de boto3 para S3
3. S3 Presigned URLs — Guía de URLs temporales
4. S3 Lifecycle Configuration — Gestión de lifecycle
5. S3 Storage Classes — Comparativa de clases de almacenamiento
6. S3 Pricing — Pricing detallado de S3
7. S3 Event Notifications — Triggers S3 → Lambda
8. S3 Best Practices — Optimización de rendimiento