Módulo 6: Cloud Migration Patterns

2. Environment Abstraction

Siguiente →

Descripción

En esta cápsula vas a aprender el patrón más fundamental de la migración cloud: environment abstraction. Es la capacidad de que tu código interactúe con LocalStack o AWS sin saber cuál es cuál. El código no pregunta "¿estoy en desarrollo o producción?" — recibe configuración y ejecuta. La diferencia entre un desarrollador junior que hardcodea endpoint_url="http://localhost:4566" y un ingeniero senior que abstrae el entorno es exactamente este patrón.

Contexto: En los Módulos 4 y 5, tu código tiene endpoint_url hardcodeado en algunos archivos y ausente en otros. Cambiar de LocalStack a AWS requiere editar líneas de código. Eso funciona para un proyecto personal, pero en un equipo de ingeniería con 3+ entornos (dev, staging, prod), no escala. Un error en un endpoint_url te conecta al entorno equivocado — y escribes datos de test en producción. Environment abstraction elimina esa clase de errores por diseño.

El Problema: Código Acoplado al Entorno

Caso real: el código del M4 y M5

Revisa cómo se crean clientes boto3 en los módulos anteriores:

# Módulo 4 — código acoplado a LocalStack
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",
)

bucket = "ai-assets-dev"
s3.put_object(Bucket=bucket, Key="test.txt", Body=b"hello")
# Módulo 5 — código acoplado a AWS
import boto3

s3 = boto3.client("s3")  # Usa ~/.aws/credentials

bucket = "ai-assets-123456789012"
s3.put_object(Bucket=bucket, Key="test.txt", Body=b"hello")

Problema visible: son dos archivos diferentes para hacer lo mismo contra dos entornos. Si quieres correr los tests del M4 contra AWS, necesitas reescribir la creación del client. Si quieres testear el código del M5 en LocalStack, necesitas agregar endpoint_url.

Los 4 elementos que cambian entre entornos

┌─────────────────────────────────────────────────────┐
│              Qué cambia entre entornos               │
├──────────────────────┬──────────────────────────────┤
│ 1. Endpoint URL      │ localhost:4566 vs AWS default │
│ 2. Credenciales      │ test/test vs IAM roles        │
│ 3. Nombres de recurso│ ai-dev vs ai-prod-123456      │
│ 4. Features          │ SageMaker no en LocalStack    │
└──────────────────────┴──────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│              Qué NO cambia entre entornos            │
├─────────────────────────────────────────────────────┤
│ - Lógica de negocio (procesar documento, invocar LLM)│
│ - Estructura de S3 keys (prompts/, documents/, etc.) │
│ - Formato de request/response                        │
│ - Manejo de errores                                  │
│ - Tests funcionales                                  │
└─────────────────────────────────────────────────────┘

La regla es simple: lo que cambia entre entornos va en configuración. Lo que no cambia va en código.

El Patrón: Environment Abstraction

Nivel 1 — Variable de entorno simple

El primer nivel de abstracción es una variable ENVIRONMENT que decide cómo crear los clientes:

import boto3
import os


def get_s3_client():
    """Crea un cliente S3 configurado para el entorno actual."""
    environment = os.environ.get("ENVIRONMENT", "local")

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

    # AWS (staging o production)
    return boto3.client("s3", region_name=os.environ.get("AWS_REGION", "us-east-1"))

Esto funciona, pero tiene problemas:

  • ❌ El if/else crece con cada entorno nuevo
  • ❌ Los valores están hardcodeados dentro de la función
  • ❌ No hay validación de que ENVIRONMENT sea un valor válido
  • ❌ No es testeable (no puedes inyectar configuración)

Nivel 2 — Config dict con endpoint switching

Un paso mejor: la configuración vive en un diccionario y el client se crea a partir de ella:

import boto3
import os
from typing import Optional


ENVIRONMENT_CONFIGS = {
    "local": {
        "endpoint_url": "http://localhost:4566",
        "aws_access_key_id": "test",
        "aws_secret_access_key": "test",
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-local",
    },
    "staging": {
        "endpoint_url": None,
        "aws_access_key_id": None,  # Usa IAM role
        "aws_secret_access_key": None,
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-staging-123456",
    },
    "production": {
        "endpoint_url": None,
        "aws_access_key_id": None,
        "aws_secret_access_key": None,
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-prod-123456",
    },
}


def get_environment_config(env_name: Optional[str] = None) -> dict:
    """Retorna la configuración para el entorno especificado."""
    env = env_name or os.environ.get("ENVIRONMENT", "local")
    if env not in ENVIRONMENT_CONFIGS:
        raise ValueError(
            f"Entorno '{env}' no reconocido. "
            f"Opciones: {list(ENVIRONMENT_CONFIGS.keys())}"
        )
    return ENVIRONMENT_CONFIGS[env]


def create_s3_client(config: dict):
    """Crea un cliente S3 a partir de un dict de configuración."""
    client_kwargs = {"region_name": config["region_name"]}
    if config.get("endpoint_url"):
        client_kwargs["endpoint_url"] = config["endpoint_url"]
    if config.get("aws_access_key_id"):
        client_kwargs["aws_access_key_id"] = config["aws_access_key_id"]
        client_kwargs["aws_secret_access_key"] = config["aws_secret_access_key"]
    return boto3.client("s3", **client_kwargs)

Mejoras:

  • ✅ Los valores no están en el código de creación del client
  • ✅ Agregar un entorno nuevo es agregar una entrada al dict
  • ✅ El client se crea a partir de config, no de lógica

Pero sigue teniendo problemas:

  • ❌ La config está en el código fuente (no en archivos .env)
  • ❌ No hay tipado ni validación
  • ❌ Los secrets están visibles en el dict

Nivel 3 — El patrón completo (lo que construirás)

El patrón maduro usa Pydantic Settings para tipar y validar, carga desde .env files, y un factory crea clients:

from pydantic_settings import BaseSettings
from typing import Optional
import boto3


class EnvironmentSettings(BaseSettings):
    """Configuración tipada del entorno."""
    environment: str = "local"
    aws_region: str = "us-east-1"
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    s3_bucket: str = "ai-assets-local"

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


def create_boto3_client(service: str, settings: EnvironmentSettings):
    """Crea un cliente boto3 configurado según el entorno."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key
    return boto3.client(service, **kwargs)

Con archivos .env por entorno:

# .env.local
ENVIRONMENT=local
AWS_ENDPOINT_URL=http://localhost:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-local
# .env.staging
ENVIRONMENT=staging
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-staging-123456
# Credenciales via IAM role, no en .env
# .env.production
ENVIRONMENT=production
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-prod-123456
# Credenciales via IAM role, no en .env

Ahora el código de negocio es completamente agnóstico:

settings = EnvironmentSettings()  # Lee de .env
s3 = create_boto3_client("s3", settings)

def process_document(doc_key: str) -> dict:
    """Procesa un documento — funciona en cualquier entorno."""
    response = s3.get_object(Bucket=settings.s3_bucket, Key=doc_key)
    content = response["Body"].read().decode("utf-8")
    # ... lógica de negocio idéntica ...
    return {"status": "processed", "key": doc_key}

Implementación Paso a Paso

Paso 1: Crear la clase de configuración

"""config/environment.py — Configuración de entorno."""

from pydantic_settings import BaseSettings
from pydantic import field_validator
from typing import Optional
from enum import Enum


class EnvironmentName(str, Enum):
    LOCAL = "local"
    STAGING = "staging"
    PRODUCTION = "production"


class EnvironmentSettings(BaseSettings):
    """Configuración del entorno con validación."""

    environment: EnvironmentName = EnvironmentName.LOCAL
    aws_region: str = "us-east-1"
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None

    s3_bucket: str = "ai-assets-local"
    lambda_function_name: str = "ai-processor-local"

    @field_validator("environment", mode="before")
    @classmethod
    def validate_environment(cls, v):
        if isinstance(v, str):
            v = v.lower().strip()
        return v

    @property
    def is_local(self) -> bool:
        return self.environment == EnvironmentName.LOCAL

    @property
    def is_aws(self) -> bool:
        return self.environment in (EnvironmentName.STAGING, EnvironmentName.PRODUCTION)

    @property
    def is_production(self) -> bool:
        return self.environment == EnvironmentName.PRODUCTION

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        use_enum_values = True

Paso 2: Función para crear clientes genéricos

"""clients/base.py — Creación de clientes boto3 agnósticos al entorno."""

import boto3
from config.environment import EnvironmentSettings


def create_client(service: str, settings: EnvironmentSettings):
    """Crea un cliente boto3 configurado para el entorno actual.

    En local (LocalStack): usa endpoint_url y credenciales test.
    En AWS (staging/prod): usa IAM role (sin endpoint_url).
    """
    kwargs = {"region_name": settings.aws_region}

    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url

    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    return boto3.client(service, **kwargs)


def create_s3_client(settings: EnvironmentSettings):
    return create_client("s3", settings)


def create_lambda_client(settings: EnvironmentSettings):
    return create_client("lambda", settings)

Paso 3: Verificar que funciona en ambos entornos

"""verify_abstraction.py — Verifica que el mismo código funciona en ambos entornos."""

import os
from config.environment import EnvironmentSettings
from clients.base import create_s3_client


def verify_environment():
    """Verifica conectividad al entorno configurado."""
    settings = EnvironmentSettings()
    print(f"Entorno: {settings.environment}")
    print(f"Bucket: {settings.s3_bucket}")
    print(f"Endpoint: {settings.aws_endpoint_url or 'AWS default'}")
    print(f"Region: {settings.aws_region}")
    print()

    s3 = create_s3_client(settings)

    try:
        s3.head_bucket(Bucket=settings.s3_bucket)
        print(f"✅ Bucket '{settings.s3_bucket}' accesible")
    except s3.exceptions.ClientError as e:
        error_code = e.response["Error"]["Code"]
        if error_code == "404":
            print(f"⚠️ Bucket '{settings.s3_bucket}' no existe. Creándolo...")
            s3.create_bucket(Bucket=settings.s3_bucket)
            print(f"✅ Bucket creado")
        else:
            print(f"❌ Error: {e}")
            return False
    except Exception as e:
        print(f"❌ No se puede conectar al entorno: {e}")
        return False

    test_key = "_health/connection-test.txt"
    s3.put_object(
        Bucket=settings.s3_bucket,
        Key=test_key,
        Body=b"connection test OK",
    )
    response = s3.get_object(Bucket=settings.s3_bucket, Key=test_key)
    content = response["Body"].read().decode("utf-8")
    assert content == "connection test OK"
    s3.delete_object(Bucket=settings.s3_bucket, Key=test_key)
    print(f"✅ Read/write verificado en '{settings.s3_bucket}'")

    return True


if __name__ == "__main__":
    verify_environment()

Ejecución:

# Contra LocalStack
ENVIRONMENT=local python verify_abstraction.py
# Output:
# Entorno: local
# Bucket: ai-assets-local
# Endpoint: http://localhost:4566
# ✅ Bucket 'ai-assets-local' accesible
# ✅ Read/write verificado

# Contra AWS (si tienes cuenta)
ENVIRONMENT=staging python verify_abstraction.py
# Output:
# Entorno: staging
# Bucket: ai-assets-staging-123456
# Endpoint: AWS default
# ✅ Bucket 'ai-assets-staging-123456' accesible
# ✅ Read/write verificado

Mismo script, mismo código, diferentes entornos. La variable ENVIRONMENT y el archivo .env correspondiente hacen toda la diferencia.

El Patrón Transparente: boto3 No Necesita Saber

Lo que boto3 realmente necesita

boto3 opera con un principio simple: si le das endpoint_url, habla con ese endpoint. Si no, habla con AWS real. Todo lo demás (credenciales, región) funciona igual:

import boto3

# Estos dos clients son funcionalmente equivalentes
# para las operaciones de S3:

# Client LocalStack
local = boto3.client("s3", endpoint_url="http://localhost:4566",
                     aws_access_key_id="test",
                     aws_secret_access_key="test")

# Client AWS
aws = boto3.client("s3")

# La misma operación funciona con ambos:
# local.put_object(Bucket="b", Key="k", Body=b"data")
# aws.put_object(Bucket="b", Key="k", Body=b"data")

El endpoint switching es el mecanismo que hace posible toda la abstracción. No necesitas wrappers complejos ni abstracciones enterprise — solo controlar qué parámetros recibe boto3.client().

Patrón completo: lógica de negocio agnóstica

"""services/document_processor.py — Lógica de negocio 100% agnóstica al entorno."""

import json
from datetime import datetime


class DocumentProcessor:
    """Procesa documentos AI. No sabe ni le importa el entorno."""

    def __init__(self, s3_client, bucket_name: str):
        self.s3 = s3_client
        self.bucket = bucket_name

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

    def store_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 save_response(self, request_id: str, result: dict) -> str:
        now = datetime.utcnow()
        key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
        payload = {
            "request_id": request_id,
            "timestamp": now.isoformat(),
            "result": result,
        }
        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 process(self, prompt_name: str, prompt_version: str, document: dict) -> dict:
        template = self.get_prompt_template(prompt_name, prompt_version)
        stored_key = self.store_document(
            "inbox", document.get("id", "unknown"), document
        )
        result = {
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "document_key": stored_key,
            "processed": True,
            "summary": f"Procesado con template: {template[:50]}...",
        }
        response_key = self.save_response(
            f"req-{datetime.utcnow().strftime('%H%M%S')}", result
        )
        result["response_key"] = response_key
        return result

Uso en cualquier entorno:

from config.environment import EnvironmentSettings
from clients.base import create_s3_client
from services.document_processor import DocumentProcessor

settings = EnvironmentSettings()
s3 = create_s3_client(settings)
processor = DocumentProcessor(s3, settings.s3_bucket)

result = processor.process(
    prompt_name="summarizer",
    prompt_version="v1",
    document={"id": "doc-001", "content": "Texto del documento..."},
)
print(result)

DocumentProcessor no importa os, no lee variables de entorno, no tiene if local/aws. Recibe un client S3 y un bucket name — y trabaja.

Troubleshooting

Problema 1: "Could not connect to the endpoint URL" con LocalStack

LocalStack no está corriendo o el puerto es diferente.

# Verificar que LocalStack está corriendo
docker ps | grep localstack

# Si no aparece:
docker run -d --name localstack -p 4566:4566 localstack/localstack

# Verificar conectividad
curl http://localhost:4566/_localstack/health

Problema 2: El client se conecta a AWS cuando debería ir a LocalStack

La variable ENVIRONMENT no está seteada o el .env no se carga.

import os
print(f"ENVIRONMENT = {os.environ.get('ENVIRONMENT', 'NO DEFINIDO')}")

# Si está vacío, Pydantic Settings usa el default ("local")
# Pero si tienes AWS credentials en ~/.aws, boto3 podría usar AWS igualmente

# Solución: asegúrate de que .env.local tiene AWS_ENDPOINT_URL

Problema 3: "Invalid endpoint" al cambiar de local a staging

El .env de staging tiene AWS_ENDPOINT_URL seteado (no debería).

# .env.staging — CORRECTO
ENVIRONMENT=staging
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-staging-123456
# Sin AWS_ENDPOINT_URL → boto3 usa AWS default

# .env.staging — INCORRECTO
ENVIRONMENT=staging
AWS_ENDPOINT_URL=http://localhost:4566  # ← Esto conecta a LocalStack, no AWS

Problema 4: Pydantic no carga el archivo .env correcto

Pydantic Settings carga .env por default. Para cargar .env.local o .env.staging, necesitas especificar:

from pydantic_settings import BaseSettings
import os

env_name = os.environ.get("ENVIRONMENT", "local")

class Settings(BaseSettings):
    class Config:
        env_file = f".env.{env_name}"

O usa la variable ENVIRONMENT directamente como env var del sistema, y el .env solo para valores adicionales.

Ejercicios Prácticos

Ejercicio 1: Multi-service client factory

Crea una función create_clients que reciba un EnvironmentSettings y retorne un diccionario con clients para S3, Lambda, y CloudWatch Logs — todos configurados para el mismo entorno.

Ver solución 
import boto3
from config.environment import EnvironmentSettings


def create_clients(settings: EnvironmentSettings) -> dict:
    """Crea clients para múltiples servicios AWS en el entorno actual."""
    services = ["s3", "lambda", "logs"]
    clients = {}

    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    for service in services:
        clients[service] = boto3.client(service, **kwargs)

    return clients


settings = EnvironmentSettings()
clients = create_clients(settings)

print(f"Entorno: {settings.environment}")
print(f"Clients creados: {list(clients.keys())}")

buckets = clients["s3"].list_buckets()
print(f"Buckets accesibles: {len(buckets.get('Buckets', []))}")

Ejercicio 2: Environment health check

Implementa una función check_environment_health que verifique la conectividad a S3 y Lambda para el entorno actual, retornando un dict con el status de cada servicio.

Ver solución 
import boto3
from config.environment import EnvironmentSettings


def check_environment_health(settings: EnvironmentSettings) -> dict:
    """Verifica la salud de los servicios en el entorno actual."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    health = {
        "environment": settings.environment,
        "services": {},
        "overall": "healthy",
    }

    # Check S3
    try:
        s3 = boto3.client("s3", **kwargs)
        s3.list_buckets()
        health["services"]["s3"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["s3"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    # Check Lambda
    try:
        lam = boto3.client("lambda", **kwargs)
        lam.list_functions(MaxItems=1)
        health["services"]["lambda"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["lambda"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    # Check bucket access
    try:
        s3 = boto3.client("s3", **kwargs)
        s3.head_bucket(Bucket=settings.s3_bucket)
        health["services"]["s3_bucket"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["s3_bucket"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    return health


settings = EnvironmentSettings()
result = check_environment_health(settings)
print(f"Entorno: {result['environment']}")
print(f"Estado general: {result['overall']}")
for service, info in result["services"].items():
    status_icon = "✅" if info["status"] == "healthy" else "❌"
    print(f"  {status_icon} {service}: {info['status']}")

Ejercicio 3: Environment comparison report

Crea un script que conecte a dos entornos (local y staging) y compare qué buckets y funciones Lambda existen en cada uno, generando un reporte de diferencias.

Ver solución 
import boto3
from config.environment import EnvironmentSettings, EnvironmentName


def get_environment_resources(settings: EnvironmentSettings) -> dict:
    """Lista los recursos disponibles en un entorno."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    resources = {"buckets": [], "functions": []}

    try:
        s3 = boto3.client("s3", **kwargs)
        response = s3.list_buckets()
        resources["buckets"] = [b["Name"] for b in response.get("Buckets", [])]
    except Exception as e:
        resources["buckets_error"] = str(e)

    try:
        lam = boto3.client("lambda", **kwargs)
        response = lam.list_functions()
        resources["functions"] = [
            f["FunctionName"] for f in response.get("Functions", [])
        ]
    except Exception as e:
        resources["functions_error"] = str(e)

    return resources


def compare_environments():
    """Compara recursos entre entorno local y staging."""
    local_settings = EnvironmentSettings(
        environment=EnvironmentName.LOCAL,
        aws_endpoint_url="http://localhost:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        s3_bucket="ai-assets-local",
    )

    local_res = get_environment_resources(local_settings)

    print("=" * 60)
    print("ENVIRONMENT COMPARISON REPORT")
    print("=" * 60)

    print(f"\n📍 LOCAL (LocalStack)")
    print(f"  Buckets: {local_res['buckets'] or 'ninguno'}")
    print(f"  Functions: {local_res['functions'] or 'ninguna'}")

    # Para staging, necesitarías credenciales AWS reales
    # Aquí mostramos el patrón:
    print(f"\n📍 STAGING (AWS)")
    print(f"  (Requiere credenciales AWS configuradas)")

    print(f"\n📊 DIFERENCIAS:")
    print(f"  Recursos solo en local: {len(local_res['buckets'])} buckets")
    print(f"  El reporte completo requiere acceso a ambos entornos")


compare_environments()

Ejercicio 4: Migración de assets entre entornos

Implementa una función que copie todos los prompt templates de un entorno (LocalStack) a otro (AWS staging), verificando que cada template se copió correctamente.

Ver solución 
import boto3
from config.environment import EnvironmentSettings, EnvironmentName


def migrate_prompts(
    source_settings: EnvironmentSettings,
    target_settings: EnvironmentSettings,
    dry_run: bool = True,
) -> dict:
    """Migra prompt templates de un entorno a otro."""

    def make_client(settings):
        kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = settings.aws_secret_access_key
        return boto3.client("s3", **kwargs)

    source_s3 = make_client(source_settings)
    target_s3 = make_client(target_settings)

    paginator = source_s3.get_paginator("list_objects_v2")
    results = {"migrated": [], "failed": [], "skipped": []}

    for page in paginator.paginate(
        Bucket=source_settings.s3_bucket, Prefix="prompts/"
    ):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if dry_run:
                results["skipped"].append(key)
                print(f"  [DRY RUN] Migraría: {key}")
                continue

            try:
                response = source_s3.get_object(
                    Bucket=source_settings.s3_bucket, Key=key
                )
                body = response["Body"].read()

                target_s3.put_object(
                    Bucket=target_settings.s3_bucket,
                    Key=key,
                    Body=body,
                    ContentType=response.get("ContentType", "application/octet-stream"),
                )

                # Verificar
                check = target_s3.get_object(
                    Bucket=target_settings.s3_bucket, Key=key
                )
                check_body = check["Body"].read()

                if body == check_body:
                    results["migrated"].append(key)
                    print(f"  ✅ Migrado y verificado: {key}")
                else:
                    results["failed"].append(key)
                    print(f"  ❌ Contenido no coincide: {key}")

            except Exception as e:
                results["failed"].append(key)
                print(f"  ❌ Error migrando {key}: {e}")

    print(f"\nResultado: {len(results['migrated'])} migrados, "
          f"{len(results['failed'])} fallidos, "
          f"{len(results['skipped'])} skipped (dry_run)")
    return results


source = EnvironmentSettings(
    environment=EnvironmentName.LOCAL,
    aws_endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    s3_bucket="ai-assets-local",
)

# Para migración real, target sería AWS staging
# target = EnvironmentSettings(environment="staging", s3_bucket="ai-assets-staging-xxx")

# Dry run contra el mismo entorno (para demo)
migrate_prompts(source, source, dry_run=True)

Resumen

  • Environment abstraction es el patrón más fundamental de migración cloud. Sin él, cada cambio de entorno requiere editar código fuente — insostenible en equipos y peligroso en producción.
  • Lo que cambia entre entornos va en configuración, lo que no cambia va en código. Endpoints, credenciales, bucket names → config. Lógica de negocio → código.
  • Tres niveles de madurez: Variable simple → config dict → Pydantic Settings con .env files. Este módulo te lleva al nivel 3.
  • boto3 es naturalmente abstracto. Solo necesita endpoint_url para LocalStack; sin él, habla con AWS. El endpoint switching es todo lo que necesitas.
  • DocumentProcessor es el ejemplo a seguir. Recibe un client y un bucket name. No sabe el entorno. No le importa.
  • En la siguiente cápsula, profundizarás en config management — Pydantic Settings, validación, secrets, y archivos .env por entorno.

Recursos Adicionales

  1. The Twelve-Factor App — Config — El principio: config en el entorno, no en el código
  2. boto3 Configuration — Cómo boto3 resuelve configuración
  3. Pydantic Settings — Settings management tipado
  4. LocalStack Configuration — Configurar LocalStack
  5. AWS SDK Configuration — Referencia de configuración AWS SDK
  6. Environment Variables Best Practices — Principios de configuración external
← Anterior1. Introducción: Cloud Migration PatternsSiguiente →3. Config Management Multi-Entorno