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:

  1. Corra contra LocalStack con ENVIRONMENT=local sin cambiar código
  2. Corra contra AWS con ENVIRONMENT=staging o ENVIRONMENT=production sin cambiar código
  3. Tenga config management tipado con Pydantic Settings y archivos .env por entorno
  4. Use dependency injection: un factory crea todos los boto3 clients según el entorno
  5. Incluya feature flags: SageMaker habilitado solo en AWS, graceful fallback en local
  6. Implemente graceful degradation: circuit breakers y fallbacks para S3 y Lambda
  7. Tenga un test suite que pase contra LocalStack Y contra AWS
  8. Incluya un migration runbook operativo: pasos concretos para migrar de local a AWS
  9. Tenga un health endpoint que reporte nivel de degradación y features disponibles

Recap del Módulo

CápsulaConceptoLo usas en el proyecto
02Environment AbstractionCódigo agnóstico al entorno en todo el proyecto
03Config ManagementPydantic Settings, .env files, validación
04Dependency InjectionClientFactory, ServiceContainer
05Testing Multi-Entornoconftest.py, markers, tests universales y por entorno
06Feature FlagsSageMaker flag, execute_if_enabled
07Graceful DegradationCircuit 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:

  1. Cambiar ENVIRONMENT=local → app vuelve a LocalStack
  2. No eliminar recursos AWS (para investigar)
  3. Revisar logs: ENVIRONMENT=staging python -c "from clients.factory import ..."
  4. 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

  1. AWS Cloud Migration Best Practices — Guía oficial de migración a AWS
  2. boto3 Documentation — SDK de Python para AWS
  3. LocalStack Documentation — Desarrollo local de servicios AWS
  4. Pydantic Settings — Config management tipado
  5. AWS Well-Architected Framework — Mejores prácticas de arquitectura
  6. Migration Strategies for Python Applications — Patrones de migración Lambda