Módulo 4: LocalStack — AWS Local Development

6. Environment Switching

Descripción

Esta cápsula te enseña la skill más valiosa de trabajar con LocalStack: escribir código que funciona contra LocalStack en desarrollo y contra AWS real en producción — cambiando solo una variable de entorno. Sin if/else de entorno en tu handler. Sin código duplicado. Sin lógica condicional de deployment. El mismo boto3, el mismo código, diferente target.

Contexto: En las cápsulas anteriores construiste un pipeline S3 + Lambda que funciona en LocalStack. Ahora te preguntas: "¿cómo paso esto a AWS real sin reescribir?" La respuesta es environment switching — y es exactamente lo que enseña esta cápsula. Este concepto es la base del Módulo 6 (Cloud Migration Patterns), donde migrarás de LocalStack a AWS con confianza.


El Concepto Core

Una variable lo cambia todo

import boto3
import os

endpoint_url = os.environ.get("AWS_ENDPOINT_URL")

s3 = boto3.client(
    "s3",
    endpoint_url=endpoint_url,  # None → AWS real, URL → LocalStack
    region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
)

Eso es todo. Cuando AWS_ENDPOINT_URL no existe (es None), boto3 se conecta a AWS real. Cuando existe (por ejemplo, http://localhost:4566), se conecta a LocalStack. Tu código no cambia. Ni una línea.

Por qué funciona

boto3 (el SDK de AWS) tiene un parámetro endpoint_url en todos sus clientes. Normalmente, boto3 calcula el endpoint automáticamente basado en el servicio y la región (s3.us-east-1.amazonaws.com). Pero si pasas endpoint_url, boto3 usa ese endpoint en lugar del calculado.

LocalStack expone las mismas APIs que AWS en localhost:4566. Cuando boto3 envía un CreateBucket a localhost:4566, LocalStack procesa la request igual que lo haría AWS. El formato de request, los headers, el payload — todo idéntico.

Desarrollo (LocalStack):
  boto3 → endpoint_url="http://localhost:4566" → LocalStack
  └── Misma API, mismos métodos, mismo formato

Producción (AWS):
  boto3 → endpoint_url=None → s3.us-east-1.amazonaws.com (AWS)
  └── Misma API, mismos métodos, mismo formato

Código:
  IDÉNTICO en ambos casos

Patrón 1: Variable de Entorno Simple

El más directo

# aws_config.py
import os
import boto3

def get_aws_client(service_name):
    """Crea un cliente AWS que apunta a LocalStack o AWS según la variable de entorno."""
    endpoint_url = os.environ.get("AWS_ENDPOINT_URL")

    kwargs = {
        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
    }

    if endpoint_url:
        kwargs["endpoint_url"] = endpoint_url
        kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")

    return boto3.client(service_name, **kwargs)


# Uso — idéntico sin importar el entorno
s3 = get_aws_client("s3")
lambda_client = get_aws_client("lambda")

Configuración por entorno

# .env.development (LocalStack)
AWS_ENDPOINT_URL=http://localhost:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_DEFAULT_REGION=us-east-1

# .env.production (AWS real)
# AWS_ENDPOINT_URL no se define → boto3 usa AWS real
AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=wJal...
AWS_DEFAULT_REGION=us-east-1
# .env.staging (LocalStack en CI/CD)
AWS_ENDPOINT_URL=http://localstack:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_DEFAULT_REGION=us-east-1

La clave: AWS_ENDPOINT_URL existe en dev y staging (LocalStack), no existe en producción (AWS).


Patrón 2: Config Class con Pydantic

Para aplicaciones FastAPI

# config.py
import os
from pydantic_settings import BaseSettings
from typing import Optional


class AWSConfig(BaseSettings):
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: str = "test"
    aws_secret_access_key: str = "test"
    aws_default_region: str = "us-east-1"
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"
    model_name: str = "gpt-4o-mini"

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    @property
    def environment_name(self) -> str:
        if self.aws_endpoint_url:
            return "localstack"
        return "aws"

    class Config:
        env_file = ".env"


aws_config = AWSConfig()
# aws_clients.py
import boto3
from config import aws_config


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


def get_lambda_client():
    kwargs = {"region_name": aws_config.aws_default_region}
    if aws_config.aws_endpoint_url:
        kwargs["endpoint_url"] = aws_config.aws_endpoint_url
        kwargs["aws_access_key_id"] = aws_config.aws_access_key_id
        kwargs["aws_secret_access_key"] = aws_config.aws_secret_access_key
    return boto3.client("lambda", **kwargs)
# main.py — tu app no sabe ni le importa si es LocalStack o AWS
from aws_clients import get_s3_client

s3 = get_s3_client()
s3.put_object(Bucket="ai-input", Key="test.txt", Body=b"hello")
# Funciona igual en LocalStack y AWS

Patrón 3: Factory con Logging

Para saber exactamente a dónde apunta

# aws_factory.py
import boto3
import os
import logging

logger = logging.getLogger(__name__)


class AWSClientFactory:
    """Crea clientes AWS con environment switching automático."""

    def __init__(self):
        self.endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
        self.region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
        self._log_target()

    def _log_target(self):
        if self.endpoint_url:
            logger.info(f"AWS target: LocalStack ({self.endpoint_url})")
        else:
            logger.info(f"AWS target: AWS real (region={self.region})")

    def _base_kwargs(self):
        kwargs = {"region_name": self.region}
        if self.endpoint_url:
            kwargs["endpoint_url"] = self.endpoint_url
            kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
            kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
        return kwargs

    def s3(self):
        return boto3.client("s3", **self._base_kwargs())

    def lambda_client(self):
        return boto3.client("lambda", **self._base_kwargs())

    @property
    def is_local(self):
        return self.endpoint_url is not None


# Singleton
aws = AWSClientFactory()
# Uso
from aws_factory import aws

s3 = aws.s3()
s3.create_bucket(Bucket="my-bucket")

if aws.is_local:
    print("Operando contra LocalStack — zero cost")
else:
    print("Operando contra AWS — verificar costes")

Patrón para Lambda Handlers

El handler que funciona en ambos entornos

# handler.py — funciona en LocalStack y AWS sin cambios
import json
import os
import boto3
from openai import OpenAI

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

openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))

def _get_s3():
    kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = "test"
        kwargs["aws_secret_access_key"] = "test"
    return boto3.client("s3", **kwargs)


def handler(event, context):
    s3 = _get_s3()

    # Este código es IDÉNTICO para LocalStack y AWS:
    doc = s3.get_object(Bucket="ai-input", Key=event["document_key"])
    text = doc["Body"].read().decode("utf-8")

    response = openai_client.chat.completions.create(
        model=os.environ.get("MODEL_NAME", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": "Analiza este documento."},
            {"role": "user", "content": text},
        ],
        max_tokens=500,
    )

    result = {
        "analysis": response.choices[0].message.content,
        "tokens": response.usage.total_tokens,
        "environment": "localstack" if AWS_ENDPOINT_URL else "aws",
    }

    s3.put_object(
        Bucket="ai-output",
        Key=f"results/{event['document_key'].split('/')[-1]}.json",
        Body=json.dumps(result),
    )

    return {"statusCode": 200, "body": json.dumps(result)}

Variables de entorno por entorno

# Deploy a LocalStack
awslocal lambda create-function \
  --function-name ai-processor \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    AWS_ENDPOINT_URL=http://host.docker.internal:4566,
    AWS_DEFAULT_REGION=us-east-1
  }" \
  ...

# Deploy a AWS real (futuro — Módulo 5)
aws lambda create-function \
  --function-name ai-processor \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    AWS_DEFAULT_REGION=us-east-1
  }" \
  ...
  # SIN AWS_ENDPOINT_URL → boto3 usa AWS real automáticamente

Docker Compose para Diferentes Entornos

Compose base + override por entorno

# docker-compose.yml (base — siempre se usa)
services:
  api:
    build: ./api
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      cache:
        condition: service_healthy

  cache:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
# docker-compose.localstack.yml (override para desarrollo con LocalStack)
services:
  api:
    environment:
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
    depends_on:
      localstack:
        condition: service_healthy

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
    volumes:
      - localstack_data:/var/lib/localstack
      - /var/run/docker.sock:/var/run/docker.sock
      - ./init-scripts:/etc/localstack/init/ready.d
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 15s

volumes:
  localstack_data:
# docker-compose.aws.yml (override para producción con AWS real)
services:
  api:
    environment:
      - AWS_DEFAULT_REGION=us-east-1
      # AWS_ENDPOINT_URL no se define → usa AWS real
      # Las credenciales vienen del IAM role del EC2/ECS
# Desarrollo con LocalStack
docker compose -f docker-compose.yml -f docker-compose.localstack.yml up -d

# Producción con AWS
docker compose -f docker-compose.yml -f docker-compose.aws.yml up -d

Validar que el Switch Funciona

Script de verificación

# verify_switch.py
import boto3
import os

endpoint = os.environ.get("AWS_ENDPOINT_URL")
environment = "LocalStack" if endpoint else "AWS"

print(f"Environment: {environment}")
print(f"Endpoint: {endpoint or 'AWS default'}")
print(f"Region: {os.environ.get('AWS_DEFAULT_REGION', 'us-east-1')}")

# Crear cliente
kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
if endpoint:
    kwargs["endpoint_url"] = endpoint
    kwargs["aws_access_key_id"] = "test"
    kwargs["aws_secret_access_key"] = "test"

s3 = boto3.client("s3", **kwargs)

# Probar operación
try:
    buckets = s3.list_buckets()
    print(f"Buckets: {[b['Name'] for b in buckets['Buckets']]}")
    print(f"Conexión exitosa a {environment}")
except Exception as e:
    print(f"Error conectando a {environment}: {e}")
# Probar contra LocalStack
AWS_ENDPOINT_URL=http://localhost:4566 python verify_switch.py
# Environment: LocalStack
# Buckets: ['ai-input', 'ai-output']
# Conexión exitosa a LocalStack

# Probar contra AWS (si tienes credenciales)
unset AWS_ENDPOINT_URL
python verify_switch.py
# Environment: AWS
# Buckets: [...]  (tus buckets reales)
# Conexión exitosa a AWS

Anti-patrones: Lo que NO Debes Hacer

Anti-patrón 1: if/else de entorno en el código

# MAL — no hagas esto
if os.environ.get("ENVIRONMENT") == "development":
    s3 = boto3.client("s3", endpoint_url="http://localhost:4566")
elif os.environ.get("ENVIRONMENT") == "staging":
    s3 = boto3.client("s3", endpoint_url="http://localstack:4566")
else:
    s3 = boto3.client("s3")

# BIEN — una sola línea
s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))

Anti-patrón 2: URLs hardcodeadas

# MAL
s3 = boto3.client("s3", endpoint_url="http://localhost:4566")

# BIEN
s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))

Anti-patrón 3: Código diferente para cada entorno

# MAL — handlers diferentes
# handler_local.py (para LocalStack)
# handler_aws.py (para AWS)

# BIEN — un solo handler, la config lo controla
# handler.py (funciona en ambos)

Anti-patrón 4: Ignorar el naming de buckets

# MAL — buckets hardcodeados
s3.put_object(Bucket="my-dev-bucket-123", Key="file.txt", Body=b"data")

# BIEN — buckets desde config
BUCKET = os.environ.get("S3_INPUT_BUCKET", "ai-input")
s3.put_object(Bucket=BUCKET, Key="file.txt", Body=b"data")

Ejercicios

Ejercicio 1: Implementar get_aws_client con logging

Crea una función get_aws_client que acepte el nombre del servicio, retorne el cliente boto3 correcto, y loguee a qué entorno se está conectando (LocalStack o AWS).

Ver solución
import boto3
import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("aws_client")


def get_aws_client(service_name):
    """Crea cliente AWS con environment switching y logging."""
    endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
    region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")

    kwargs = {"region_name": region}

    if endpoint_url:
        kwargs["endpoint_url"] = endpoint_url
        kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
        logger.info(f"[{service_name}] → LocalStack ({endpoint_url})")
    else:
        logger.info(f"[{service_name}] → AWS ({region})")

    return boto3.client(service_name, **kwargs)


# Test
s3 = get_aws_client("s3")
lambda_c = get_aws_client("lambda")

# Con LocalStack:
# INFO:aws_client:[s3] → LocalStack (http://localhost:4566)
# INFO:aws_client:[lambda] → LocalStack (http://localhost:4566)

Ejercicio 2: Config class con validación

Crea una clase AWSConfig con Pydantic que valide la configuración: si aws_endpoint_url está definido, las credenciales pueden ser test. Si no está definido (AWS real), las credenciales deben empezar con AKIA (access key real). Incluye una propiedad is_local.

Ver solución
from pydantic_settings import BaseSettings
from pydantic import model_validator
from typing import Optional


class AWSConfig(BaseSettings):
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: str = "test"
    aws_secret_access_key: str = "test"
    aws_default_region: str = "us-east-1"
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    @model_validator(mode="after")
    def validate_credentials(self):
        if not self.is_local:
            if not self.aws_access_key_id.startswith("AKIA"):
                raise ValueError(
                    "AWS real requiere credenciales válidas "
                    "(access key debe empezar con AKIA)"
                )
        return self

    class Config:
        env_file = ".env"


# Test con LocalStack (funciona con test/test)
import os
os.environ["AWS_ENDPOINT_URL"] = "http://localhost:4566"
config = AWSConfig()
print(f"Local: {config.is_local}")  # True

# Test sin endpoint (requiere credenciales reales)
del os.environ["AWS_ENDPOINT_URL"]
os.environ["AWS_ACCESS_KEY_ID"] = "AKIA1234567890EXAMPLE"
os.environ["AWS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
config = AWSConfig()
print(f"Local: {config.is_local}")  # False

Ejercicio 3: Docker Compose multi-entorno

Crea un docker-compose.yml base y un docker-compose.localstack.yml override. La app base tiene FastAPI. El override agrega LocalStack y configura las variables de entorno. Demuestra que el switch funciona levantando ambos.

Ver solución
# docker-compose.yml (base)
services:
  api:
    build: ./api
    ports:
      - "8000:8000"
    environment:
      - AWS_DEFAULT_REGION=us-east-1
# docker-compose.localstack.yml (override)
services:
  api:
    environment:
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
    depends_on:
      localstack:
        condition: service_healthy

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 15s
# Desarrollo con LocalStack
docker compose -f docker-compose.yml -f docker-compose.localstack.yml up -d

# Verificar que api tiene las variables
docker compose exec api env | grep AWS
# AWS_ENDPOINT_URL=http://localstack:4566
# AWS_ACCESS_KEY_ID=test

# Producción (solo base — sin LocalStack)
docker compose up -d
docker compose exec api env | grep AWS
# AWS_DEFAULT_REGION=us-east-1
# (sin AWS_ENDPOINT_URL)

Ejercicio 4: Test de compatibilidad LocalStack ↔ AWS

Escribe un script que ejecute una secuencia de operaciones S3 (crear bucket, subir archivo, descargar, listar, eliminar) y verifique que funciona tanto contra LocalStack como contra AWS. El script debe reportar success/failure para cada operación.

Ver solución
# test_compatibility.py
import boto3
import json
import os

endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
env_name = "LocalStack" if endpoint_url else "AWS"

kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
if endpoint_url:
    kwargs["endpoint_url"] = endpoint_url
    kwargs["aws_access_key_id"] = "test"
    kwargs["aws_secret_access_key"] = "test"

s3 = boto3.client("s3", **kwargs)
bucket = "compatibility-test-bucket"
key = "test/data.json"
data = {"test": True, "environment": env_name}

results = {}

# 1. Create bucket
try:
    s3.create_bucket(Bucket=bucket)
    results["create_bucket"] = "PASS"
except Exception as e:
    results["create_bucket"] = f"FAIL: {e}"

# 2. Put object
try:
    s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(data))
    results["put_object"] = "PASS"
except Exception as e:
    results["put_object"] = f"FAIL: {e}"

# 3. Get object
try:
    resp = s3.get_object(Bucket=bucket, Key=key)
    downloaded = json.loads(resp["Body"].read())
    assert downloaded == data
    results["get_object"] = "PASS"
except Exception as e:
    results["get_object"] = f"FAIL: {e}"

# 4. List objects
try:
    resp = s3.list_objects_v2(Bucket=bucket)
    assert resp["KeyCount"] == 1
    results["list_objects"] = "PASS"
except Exception as e:
    results["list_objects"] = f"FAIL: {e}"

# 5. Delete object
try:
    s3.delete_object(Bucket=bucket, Key=key)
    results["delete_object"] = "PASS"
except Exception as e:
    results["delete_object"] = f"FAIL: {e}"

# 6. Delete bucket
try:
    s3.delete_bucket(Bucket=bucket)
    results["delete_bucket"] = "PASS"
except Exception as e:
    results["delete_bucket"] = f"FAIL: {e}"

# Report
print(f"\n{'='*40}")
print(f"Compatibility Test — {env_name}")
print(f"{'='*40}")
passed = 0
for op, result in results.items():
    status = "✅" if result == "PASS" else "❌"
    print(f"  {status} {op}: {result}")
    if result == "PASS":
        passed += 1

print(f"\nResultado: {passed}/{len(results)} operaciones exitosas")
# Test contra LocalStack
AWS_ENDPOINT_URL=http://localhost:4566 python test_compatibility.py

# Test contra AWS (si tienes credenciales)
unset AWS_ENDPOINT_URL
python test_compatibility.py

Troubleshooting

"Mi código funciona en LocalStack pero falla en AWS"

# Causa más común: IAM permissions
# LocalStack Community no enforce IAM
# AWS requiere permisos explícitos

# Solución: verifica que tu role/usuario tiene permisos para S3 y Lambda
aws iam list-attached-user-policies --user-name tu-usuario

# Para S3: necesitas s3:PutObject, s3:GetObject, s3:ListBucket
# Para Lambda: necesitas lambda:InvokeFunction

"Las credenciales 'test' no funcionan en AWS"

# Correcto — 'test/test' es para LocalStack
# AWS necesita credenciales reales
aws configure
# Ingresa tus credenciales reales (AKIA...)

"endpoint_url=None causa error"

# boto3 acepta endpoint_url=None — lo ignora y usa el default
# Si tienes error, verifica que no estás pasando un string vacío:

# MAL
endpoint = os.environ.get("AWS_ENDPOINT_URL", "")
# Si AWS_ENDPOINT_URL no existe, endpoint = "" (string vacío, no None)

# BIEN
endpoint = os.environ.get("AWS_ENDPOINT_URL")
# Si no existe, endpoint = None (boto3 lo maneja correctamente)

Resumen

  • Environment switching = el mismo código apunta a LocalStack o AWS cambiando solo AWS_ENDPOINT_URL.
  • Cuando AWS_ENDPOINT_URL es None, boto3 conecta a AWS real. Cuando tiene valor, conecta a ese endpoint (LocalStack).
  • El patrón más limpio: boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL")). Una línea.
  • Docker Compose overrides permiten levantar con o sin LocalStack según el entorno.
  • Anti-patrones a evitar: if/else de entorno, URLs hardcodeadas, código duplicado por entorno.
  • Este patrón es la base de la migración en el Módulo 6: si tu código funciona con environment switching, migrar a AWS es cambiar una variable.
  • Recuerda: LocalStack Community no enforce IAM. En AWS real, verifica permisos.

Recursos Adicionales

  1. boto3 Configuration — Cómo boto3 resuelve credenciales y endpoints
  2. AWS Environment Variables — Variables de entorno oficiales de AWS
  3. 12-Factor App — Config — La filosofía de config por variables de entorno
  4. Pydantic Settings — Gestión de settings con Pydantic
  5. Docker Compose Override — Cómo funcionan los overrides en Compose
  6. LocalStack AWS Feature Coverage — Qué funciona igual y qué tiene diferencias
  7. AWS IAM Best Practices — Para cuando migres a AWS real