Módulo 6: Cloud Migration Patterns
8. Proyecto: Migration-Ready AI App
Descripción del proyecto
Este es el proyecto integrador del Módulo 6. Vas a refactorizar la app AI de los módulos anteriores en una aplicación migration-ready: el mismo código corre contra LocalStack en desarrollo y contra AWS en staging/producción sin cambiar una línea de lógica de negocio. La aplicación tiene environment abstraction, config management con Pydantic Settings, dependency injection para boto3 clients, tests multi-entorno, feature flags para SageMaker, graceful degradation con circuit breakers, y un migration runbook documentado que otro ingeniero puede seguir paso a paso.
Por qué importa: Este proyecto integra todo lo que aprendiste en las cápsulas del módulo: environment abstraction (C02), config management (C03), dependency injection (C04), testing multi-entorno (C05), feature flags (C06), y graceful degradation (C07). Es el artefacto más sofisticado de la Phase 2 y el que llevarás al Proyecto Integrador (M8). Cuando termines, tendrás una app que demuestra que sabes construir software de producción — no prototipos que "funcionan en mi máquina."
Objetivo del proyecto
Producir una Migration-Ready AI App funcional que:
- Corra contra LocalStack con
ENVIRONMENT=localsin cambiar código - Corra contra AWS con
ENVIRONMENT=stagingoENVIRONMENT=productionsin cambiar código - Tenga config management tipado con Pydantic Settings y archivos .env por entorno
- Use dependency injection: un factory crea todos los boto3 clients según el entorno
- Incluya feature flags: SageMaker habilitado solo en AWS, graceful fallback en local
- Implemente graceful degradation: circuit breakers y fallbacks para S3 y Lambda
- Tenga un test suite que pase contra LocalStack Y contra AWS
- Incluya un migration runbook operativo: pasos concretos para migrar de local a AWS
- Tenga un health endpoint que reporte nivel de degradación y features disponibles
Recap del Módulo
| Cápsula | Concepto | Lo usas en el proyecto |
|---|---|---|
| 02 | Environment Abstraction | Código agnóstico al entorno en todo el proyecto |
| 03 | Config Management | Pydantic Settings, .env files, validación |
| 04 | Dependency Injection | ClientFactory, ServiceContainer |
| 05 | Testing Multi-Entorno | conftest.py, markers, tests universales y por entorno |
| 06 | Feature Flags | SageMaker flag, execute_if_enabled |
| 07 | Graceful Degradation | Circuit breakers, fallbacks, health levels |
Especificaciones Técnicas
Arquitectura
Migration-Ready AI App
├── ENVIRONMENT=local (LocalStack) ENVIRONMENT=staging (AWS)
│ ┌──────────────────────────┐ ┌──────────────────────────┐
│ │ LocalStack :4566 │ │ AWS us-east-1 │
│ │ ├── S3 (local bucket) │ │ ├── S3 (staging bucket) │
│ │ ├── Lambda (local) │ │ ├── Lambda (staging) │
│ │ └── SageMaker: ❌ N/A │ │ ├── SageMaker: ✅ │
│ └──────────────────────────┘ │ └── CloudWatch: ✅ │
│ └──────────────────────────┘
│ ▲ ▲
│ │ │
│ └───────────┐ ┌────────────────┘
│ │ │
│ ┌───────────┴──────┴────────────┐
│ │ MISMO CÓDIGO FUENTE │
│ │ │
│ │ config/settings.py │
│ │ clients/factory.py │
│ │ services/processor.py │
│ │ services/feature_flags.py │
│ │ services/health.py │
│ │ handler.py │
│ │ tests/conftest.py │
│ └────────────────────────────────┘
Estructura de archivos
migration-ready-app/
├── config/
│ ├── __init__.py
│ ├── settings.py ← Pydantic Settings con validación
│ ├── loader.py ← get_settings() con env detection
│ └── validators.py ← Validación por entorno
├── clients/
│ ├── __init__.py
│ ├── factory.py ← ClientFactory (boto3 clients)
│ └── resilient.py ← ResilientClient wrapper
├── services/
│ ├── __init__.py
│ ├── container.py ← ServiceContainer (DI wiring)
│ ├── document_processor.py ← Lógica de negocio
│ ├── feature_flags.py ← Feature flags
│ ├── circuit_breaker.py ← Circuit breaker
│ ├── fallbacks.py ← Fallback strategies
│ └── health.py ← Health checker con degradation
├── tests/
│ ├── conftest.py ← Fixtures multi-entorno
│ ├── test_s3_operations.py ← Tests S3 universales
│ ├── test_processor.py ← Tests del processor
│ ├── test_feature_flags.py ← Tests de feature flags
│ ├── test_degradation.py ← Tests de graceful degradation
│ ├── test_aws_only.py ← Tests solo AWS
│ └── test_migration.py ← Tests de migración end-to-end
├── migration/
│ └── RUNBOOK.md ← Migration runbook paso a paso
├── .env.local ← Config LocalStack
├── .env.staging ← Config AWS staging
├── .env.production ← Config AWS producción
├── .env.example ← Template para nuevos developers
├── .gitignore ← Excluye .env files
├── handler.py ← Lambda handler (punto de entrada)
├── pytest.ini ← Configuración de pytest
└── requirements.txt ← Dependencias
Endpoints requeridos
POST /process → Procesa un documento con prompt template
GET /health → Status con niveles de degradación y features
Request/Response format
// POST /process — Request
{
"prompt_name": "summarizer",
"prompt_version": "v1",
"document": {
"id": "doc-001",
"title": "Guía de deployment",
"content": "El deployment de aplicaciones AI requiere..."
}
}
// POST /process — Response (healthy)
{
"status": "ok",
"data": {
"prompt_used": "summarizer/v1",
"document_stored": "documents/inbox/doc-001.json",
"processed": true,
"template_preview": "Resume el documento en 3 puntos...",
"sagemaker_enrichment": null,
"degraded": false,
"degradation_details": []
}
}
// POST /process — Response (degraded, S3 lento)
{
"status": "partial",
"data": {
"prompt_used": "summarizer/v1",
"document_stored": null,
"processed": true,
"template_preview": "Genera un resumen conciso...",
"degraded": true,
"degradation_details": [
"Storage: documento no persistido (CircuitBreakerError)"
]
},
"degradation": {
"level": "partial",
"affected": ["s3_storage"],
"fallbacks": ["default_prompt", "skip_persistence"]
}
}
// GET /health — Response
{
"status": "degraded",
"environment": "staging",
"message": "Servicios opcionales no disponibles: ['sagemaker']",
"available_features": ["s3", "lambda", "cloudwatch"],
"degraded_features": ["sagemaker"],
"features": {
"sagemaker_enrichment": {"enabled": true, "service_health": "unhealthy"},
"cloudwatch_metrics": {"enabled": true},
"cost_tracking": {"enabled": true}
},
"services": [
{"name": "s3", "status": "healthy", "latency_ms": 12.3, "critical": true},
{"name": "lambda", "status": "healthy", "latency_ms": 45.1, "critical": true},
{"name": "sagemaker", "status": "unhealthy", "latency_ms": 5001.2, "critical": false}
]
}
Implementación Paso a Paso
Paso 1: config/settings.py
"""config/settings.py — Settings de la Migration-Ready AI App."""
from pydantic_settings import BaseSettings
from pydantic import field_validator, model_validator
from typing import Optional
from enum import Enum
class EnvironmentName(str, Enum):
LOCAL = "local"
STAGING = "staging"
PRODUCTION = "production"
class Settings(BaseSettings):
environment: EnvironmentName = EnvironmentName.LOCAL
app_name: str = "migration-ready-ai-app"
app_version: str = "1.0.0"
debug: bool = False
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"
lambda_timeout: int = 120
lambda_memory: int = 768
openai_api_key: Optional[str] = None
openai_model: str = "gpt-4o-mini"
openai_max_tokens: int = 1000
feature_sagemaker_enabled: bool = False
feature_advanced_logging: bool = False
feature_cost_tracking: bool = False
log_level: str = "INFO"
max_retries: int = 3
circuit_breaker_threshold: int = 5
circuit_breaker_reset: int = 60
@field_validator("environment", mode="before")
@classmethod
def normalize_env(cls, v):
return v.lower().strip() if isinstance(v, str) else v
@model_validator(mode="after")
def validate_config(self):
if self.environment == EnvironmentName.LOCAL:
if not self.aws_endpoint_url:
self.aws_endpoint_url = "http://localhost:4566"
if not self.aws_access_key_id:
self.aws_access_key_id = "test"
self.aws_secret_access_key = "test"
if self.environment == EnvironmentName.PRODUCTION:
if self.debug:
raise ValueError("debug=True no permitido en producción")
if self.aws_endpoint_url:
raise ValueError("aws_endpoint_url no debe setearse en producción")
return self
@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)
class Config:
env_file = ".env"
env_file_encoding = "utf-8"
use_enum_values = True
Paso 2: clients/factory.py
"""clients/factory.py — Factory de boto3 clients."""
import boto3
from typing import Any
from config.settings import Settings
class ClientFactory:
def __init__(self, settings: Settings):
self.settings = settings
self._kwargs = self._build_kwargs()
self._clients: dict[str, Any] = {}
def _build_kwargs(self) -> dict:
kwargs = {"region_name": self.settings.aws_region}
if self.settings.aws_endpoint_url:
kwargs["endpoint_url"] = self.settings.aws_endpoint_url
if self.settings.aws_access_key_id:
kwargs["aws_access_key_id"] = self.settings.aws_access_key_id
kwargs["aws_secret_access_key"] = self.settings.aws_secret_access_key
return kwargs
def get_client(self, service: str) -> Any:
if service not in self._clients:
self._clients[service] = boto3.client(service, **self._kwargs)
return self._clients[service]
@property
def s3(self) -> Any:
return self.get_client("s3")
@property
def lambda_client(self) -> Any:
return self.get_client("lambda")
def health_check(self) -> dict:
results = {}
for service, client in self._clients.items():
try:
if service == "s3":
client.list_buckets()
elif service == "lambda":
client.list_functions(MaxItems=1)
results[service] = "healthy"
except Exception as e:
results[service] = f"unhealthy: {e}"
return results
Paso 3: services/container.py
"""services/container.py — Service container con DI completo."""
from dataclasses import dataclass
from config.settings import Settings
from clients.factory import ClientFactory
from services.document_processor import ResilientDocumentProcessor
from services.feature_flags import FeatureFlags
from services.health import HealthChecker
@dataclass
class ServiceContainer:
settings: Settings
factory: ClientFactory
flags: FeatureFlags
processor: ResilientDocumentProcessor
health_checker: HealthChecker
@classmethod
def create(cls, settings: Settings | None = None) -> "ServiceContainer":
if settings is None:
import os
env = os.environ.get("ENVIRONMENT", "local")
env_file = f".env.{env}"
if os.path.exists(env_file):
settings = Settings(_env_file=env_file)
else:
settings = Settings()
factory = ClientFactory(settings)
flags = FeatureFlags(settings)
sagemaker_client = None
if flags.is_enabled("sagemaker_enrichment"):
try:
sagemaker_client = factory.get_client("sagemaker-runtime")
except Exception:
pass
processor = ResilientDocumentProcessor(
s3_client=factory.s3,
bucket=settings.s3_bucket,
feature_flags=flags,
sagemaker_client=sagemaker_client,
max_retries=settings.max_retries,
circuit_threshold=settings.circuit_breaker_threshold,
circuit_reset=settings.circuit_breaker_reset,
)
health_checker = HealthChecker(factory, settings, flags)
return cls(
settings=settings,
factory=factory,
flags=flags,
processor=processor,
health_checker=health_checker,
)
Paso 4: handler.py
"""handler.py — Lambda handler de la Migration-Ready AI App."""
import json
import logging
from services.container import ServiceContainer
logger = logging.getLogger(__name__)
container: ServiceContainer | None = None
def get_container() -> ServiceContainer:
global container
if container is None:
container = ServiceContainer.create()
logger.info(
f"Container inicializado: env={container.settings.environment}"
)
return container
def lambda_handler(event, context):
c = get_container()
path = event.get("rawPath", event.get("path", ""))
method = event.get("requestContext", {}).get("http", {}).get("method", "GET")
if path == "/health" and method == "GET":
return handle_health(c)
if path == "/process" and method == "POST":
return handle_process(c, event, context)
return {"statusCode": 404, "body": json.dumps({"error": "Not found"})}
def handle_health(c: ServiceContainer) -> dict:
health = c.health_checker.check()
features = c.flags.status()
return {
"statusCode": 200 if health.level.value in ("healthy", "degraded") else 503,
"body": json.dumps({
"status": health.level.value,
"environment": c.settings.environment,
"message": health.message,
"available_features": health.available_features,
"degraded_features": health.degraded_features,
"features": features,
"services": [
{
"name": s.name,
"status": s.status,
"latency_ms": round(s.latency_ms, 1),
"critical": s.critical,
}
for s in health.services
],
}),
}
def handle_process(c: ServiceContainer, event: dict, context) -> dict:
try:
body = json.loads(event.get("body", "{}"))
except json.JSONDecodeError:
return {
"statusCode": 400,
"body": json.dumps({"error": "Invalid JSON in request body"}),
}
prompt_name = body.get("prompt_name", "summarizer")
prompt_version = body.get("prompt_version", "v1")
document = body.get("document", {})
if not document:
return {
"statusCode": 400,
"body": json.dumps({"error": "document is required"}),
}
result = c.processor.process(prompt_name, prompt_version, document)
status = "ok" if not result.get("degraded") else "partial"
response_body = {"status": status, "data": result}
if result.get("degraded"):
response_body["degradation"] = {
"level": "partial",
"affected": [d.split(":")[0] for d in result.get("degradation_details", [])],
"details": result.get("degradation_details", []),
}
return {
"statusCode": 200,
"headers": {
"Content-Type": "application/json",
"X-Environment": c.settings.environment,
"X-Degradation": "none" if not result.get("degraded") else "partial",
},
"body": json.dumps(response_body, default=str),
}
Paso 5: .env files
# .env.local
ENVIRONMENT=local
DEBUG=true
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
LAMBDA_FUNCTION_NAME=ai-processor-local
FEATURE_SAGEMAKER_ENABLED=false
FEATURE_ADVANCED_LOGGING=false
FEATURE_COST_TRACKING=false
LOG_LEVEL=DEBUG
MAX_RETRIES=3
CIRCUIT_BREAKER_THRESHOLD=5
CIRCUIT_BREAKER_RESET=30
# .env.staging
ENVIRONMENT=staging
DEBUG=true
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-staging-123456789012
LAMBDA_FUNCTION_NAME=ai-processor-staging
FEATURE_SAGEMAKER_ENABLED=true
FEATURE_ADVANCED_LOGGING=true
FEATURE_COST_TRACKING=true
LOG_LEVEL=INFO
MAX_RETRIES=3
CIRCUIT_BREAKER_THRESHOLD=5
CIRCUIT_BREAKER_RESET=60
# .env.production
ENVIRONMENT=production
DEBUG=false
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-prod-123456789012
LAMBDA_FUNCTION_NAME=ai-processor-prod
FEATURE_SAGEMAKER_ENABLED=true
FEATURE_ADVANCED_LOGGING=true
FEATURE_COST_TRACKING=true
LOG_LEVEL=WARNING
MAX_RETRIES=5
CIRCUIT_BREAKER_THRESHOLD=3
CIRCUIT_BREAKER_RESET=60
Paso 6: requirements.txt
boto3>=1.34.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
python-dotenv>=1.0.0
pytest>=7.0.0
openai>=1.0.0
Migration Runbook
RUNBOOK.md — El artefacto operativo
# Migration Runbook: LocalStack → AWS Staging
## Pre-requisitos
- [ ] Cuenta AWS con IAM user/role configurado
- [ ] AWS CLI configurado (`aws sts get-caller-identity` funciona)
- [ ] Tests pasan en LocalStack: `ENVIRONMENT=local pytest tests/ -v`
- [ ] .env.staging creado con valores correctos
## Paso 1: Verificar config de staging
```bash
# Validar que .env.staging tiene los valores correctos
python -c "
from config.settings import Settings
s = Settings(_env_file='.env.staging')
print(f'Entorno: {s.environment}')
print(f'Bucket: {s.s3_bucket}')
print(f'SageMaker: {s.feature_sagemaker_enabled}')
assert s.environment == 'staging'
assert 'localhost' not in (s.aws_endpoint_url or '')
print('✅ Config de staging válida')
"
Paso 2: Crear recursos AWS
# Crear bucket S3
aws s3 mb s3://ai-assets-staging-123456789012
# Verificar
aws s3 ls s3://ai-assets-staging-123456789012
Paso 3: Migrar prompt templates
# Exportar prompts de LocalStack
ENVIRONMENT=local python -c "
from clients.factory import ClientFactory
from config.settings import Settings
s = Settings(_env_file='.env.local')
f = ClientFactory(s)
# ... export logic
"
# Importar a AWS staging
ENVIRONMENT=staging python -c "
# ... import logic
"
Paso 4: Correr tests contra staging
ENVIRONMENT=staging pytest tests/ -v --tb=short
# Esperado:
# - Tests universales: PASSED
# - Tests aws_only: PASSED
# - Tests local_only: SKIPPED
# - Tests sagemaker: PASSED (si endpoint desplegado)
Paso 5: Verificar health endpoint
ENVIRONMENT=staging python -c "
from services.container import ServiceContainer
c = ServiceContainer.create()
health = c.health_checker.check()
print(f'Status: {health.level.value}')
for s in health.services:
icon = '✅' if s.status == 'healthy' else '❌'
print(f' {icon} {s.name}: {s.status} ({s.latency_ms:.0f}ms)')
"
Paso 6: Smoke test funcional
ENVIRONMENT=staging python -c "
from services.container import ServiceContainer
c = ServiceContainer.create()
result = c.processor.process(
'summarizer', 'v1',
{'id': 'smoke-test', 'content': 'Test de migración'}
)
print(f'Resultado: {result}')
assert result['processed'] == True
print('✅ Smoke test pasó')
"
Rollback
Si algo falla:
- Cambiar
ENVIRONMENT=local→ app vuelve a LocalStack - No eliminar recursos AWS (para investigar)
- Revisar logs:
ENVIRONMENT=staging python -c "from clients.factory import ..." - Comparar config:
python config/compare.py local staging
Verificación Final
-
ENVIRONMENT=local pytest→ todos los tests pasan -
ENVIRONMENT=staging pytest→ todos los tests pasan - Health endpoint retorna "healthy" o "degraded" (no "unavailable")
- Smoke test funcional pasa
- Feature flags reportan correctamente
- Rollback verificado (cambiar a local y verificar)
---
## Checklist de Entrega
### Funcionalidad
- [ ] La app corre con `ENVIRONMENT=local` contra LocalStack
- [ ] La app corre con `ENVIRONMENT=staging` contra AWS (o simula exitosamente)
- [ ] `POST /process` procesa documentos y retorna resultado
- [ ] `GET /health` reporta servicios, features, y nivel de degradación
- [ ] Feature flags habilitan/deshabilitan SageMaker correctamente
- [ ] Circuit breaker protege contra fallos de S3
- [ ] Fallback retorna prompts default cuando S3 no responde
### Arquitectura
- [ ] `config/settings.py` tiene Pydantic Settings con validación
- [ ] `clients/factory.py` tiene ClientFactory con cache de clients
- [ ] `services/container.py` tiene ServiceContainer con DI
- [ ] `services/feature_flags.py` tiene FeatureFlags con execute_if_enabled
- [ ] `services/circuit_breaker.py` tiene CircuitBreaker funcional
- [ ] `handler.py` usa ServiceContainer (no construye clients directamente)
### Testing
- [ ] `tests/conftest.py` detecta entorno y configura fixtures
- [ ] Tests universales (S3, processor) pasan en local
- [ ] Tests aws_only se saltan en local, pasan en staging
- [ ] Tests de degradación verifican fallbacks
- [ ] `pytest.ini` configurado con markers
### Config
- [ ] `.env.local` configurado para LocalStack
- [ ] `.env.staging` configurado para AWS
- [ ] `.env.production` configurado (aunque no se use aún)
- [ ] `.env.example` como template
- [ ] `.gitignore` excluye archivos .env
### Documentación
- [ ] `migration/RUNBOOK.md` con pasos concretos
- [ ] Runbook incluye rollback
- [ ] Runbook incluye verificación en cada paso
---
## Verificación del Proyecto
### Script de verificación automatizada
```python
"""verify_project.py — Verifica que el proyecto cumple todos los requisitos."""
import os
import sys
import importlib
def check_file_exists(path: str) -> bool:
exists = os.path.exists(path)
icon = "✅" if exists else "❌"
print(f" {icon} {path}")
return exists
def check_module_imports(module_name: str) -> bool:
try:
importlib.import_module(module_name)
print(f" ✅ import {module_name}")
return True
except Exception as e:
print(f" ❌ import {module_name}: {e}")
return False
def verify():
print("=" * 60)
print("MIGRATION-READY AI APP — VERIFICACIÓN")
print("=" * 60)
results = []
print("\n📁 Archivos requeridos:")
required_files = [
"config/settings.py",
"config/loader.py",
"clients/factory.py",
"services/container.py",
"services/document_processor.py",
"services/feature_flags.py",
"services/circuit_breaker.py",
"services/health.py",
"tests/conftest.py",
"tests/test_s3_operations.py",
"tests/test_processor.py",
"migration/RUNBOOK.md",
"handler.py",
".env.local",
".env.staging",
".env.example",
"requirements.txt",
"pytest.ini",
]
for f in required_files:
results.append(check_file_exists(f))
print("\n📦 Módulos importables:")
modules = [
"config.settings",
"clients.factory",
"services.container",
"services.feature_flags",
]
for m in modules:
results.append(check_module_imports(m))
print("\n🔧 Configuración:")
try:
from config.settings import Settings
s = Settings()
print(f" ✅ Settings carga: env={s.environment}")
results.append(True)
except Exception as e:
print(f" ❌ Settings falla: {e}")
results.append(False)
print("\n📊 Resultado:")
passed = sum(results)
total = len(results)
pct = passed / total * 100 if total > 0 else 0
print(f" {passed}/{total} checks pasaron ({pct:.0f}%)")
if pct == 100:
print("\n🎉 Proyecto completo. Listo para migrar.")
elif pct >= 80:
print("\n⚠️ Casi listo. Revisa los items que fallan.")
else:
print("\n❌ Proyecto incompleto. Revisa la checklist.")
return pct == 100
if __name__ == "__main__":
success = verify()
sys.exit(0 if success else 1)
Ejecución de verificación
# Verificar estructura
python verify_project.py
# Tests contra LocalStack
ENVIRONMENT=local pytest tests/ -v
# Tests contra AWS (si disponible)
ENVIRONMENT=staging pytest tests/ -v
# Health check
ENVIRONMENT=local python -c "
from services.container import ServiceContainer
c = ServiceContainer.create()
h = c.health_checker.check()
print(f'Health: {h.level.value} — {h.message}')
"
Conexión con Módulos Siguientes
Módulo 7: Alternative Platforms
La abstraction layer que construiste aquí facilita evaluar alternativas. Si la decision matrix del M7 dice "Render en lugar de AWS," tu app migration-ready puede adaptarse:
M6 (aquí): ENVIRONMENT=local → LocalStack
ENVIRONMENT=staging → AWS
M7: ENVIRONMENT=render → Render.com
ENVIRONMENT=railway → Railway.app
La abstraction layer soporta nuevos entornos
agregando config, no reescribiendo código.
Módulo 8: Proyecto Integrador
La app migration-ready del M6 es el artefacto que se despliega end-to-end en M8:
M6: Construye la app migration-ready (arquitectura)
↓
M7: Evalúa plataformas alternativas (decisión)
↓
M8: Despliega a producción con CI/CD + monitoring (operación)
├── La app del M6 se despliega
├── CI/CD usa los tests del M6
├── Monitoring usa el health check del M6
└── Migration runbook se ejecuta en producción
Lo que llevas al M8
- ✅ App con environment abstraction (corre en cualquier entorno)
- ✅ Config management que soporta múltiples entornos
- ✅ Test suite que verifica migración
- ✅ Health endpoint con degradation levels
- ✅ Feature flags para capacidades opcionales
- ✅ Migration runbook documentado
Criterios de Evaluación
Nivel Básico (aprobado)
- La app corre contra LocalStack con
ENVIRONMENT=local - Config management con Pydantic Settings y .env files
- ClientFactory crea clients según el entorno
- Al menos 10 tests pasan contra LocalStack
- Migration runbook existe con pasos concretos
Nivel Intermedio (bien hecho)
- Todo lo básico +
- Feature flags habilitan/deshabilitan features por entorno
- Circuit breaker protege contra fallos de S3
- Health endpoint reporta servicios y features
- Tests con markers (
aws_only,local_only) - 20+ tests pasan
Nivel Avanzado (excelente)
- Todo lo intermedio +
- Graceful degradation completa con fallbacks en cascada
- Tests de degradación que simulan fallos de servicio
- Config validation que rechaza configuraciones inválidas
- Dynamic feature flags (actualizables en runtime)
- La app pasa tests contra LocalStack Y contra AWS
- 30+ tests con cobertura de todos los patrones del módulo
Resumen
- Este proyecto es el artefacto más completo de la Phase 2. Integra environment abstraction, config management, dependency injection, feature flags, graceful degradation, y testing multi-entorno.
- Mismo código, cualquier entorno. Cambiar de LocalStack a AWS es cambiar una variable de entorno. El código, los tests, y el health check se adaptan automáticamente.
- El migration runbook es parte del entregable. No es solo código — es documentación operativa que otro ingeniero puede seguir.
- Este artefacto se lleva al M8. La app migration-ready es lo que se despliega a producción con CI/CD, monitoring, y operación real en el Proyecto Integrador.
- Los patrones son transferibles. Abstraction, DI, feature flags, circuit breakers — estos patrones aplican a cualquier sistema, no solo a LocalStack/AWS. Son skills de carrera.
Recursos Adicionales
- AWS Cloud Migration Best Practices — Guía oficial de migración a AWS
- boto3 Documentation — SDK de Python para AWS
- LocalStack Documentation — Desarrollo local de servicios AWS
- Pydantic Settings — Config management tipado
- AWS Well-Architected Framework — Mejores prácticas de arquitectura
- Migration Strategies for Python Applications — Patrones de migración Lambda