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_URLesNone, 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/elsede 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
- boto3 Configuration — Cómo boto3 resuelve credenciales y endpoints
- AWS Environment Variables — Variables de entorno oficiales de AWS
- 12-Factor App — Config — La filosofía de config por variables de entorno
- Pydantic Settings — Gestión de settings con Pydantic
- Docker Compose Override — Cómo funcionan los overrides en Compose
- LocalStack AWS Feature Coverage — Qué funciona igual y qué tiene diferencias
- AWS IAM Best Practices — Para cuando migres a AWS real