Módulo 6: Cloud Migration Patterns

6. Feature Flags para Cloud

Siguiente →

Descripción

En esta cápsula vas a implementar feature flags para manejar servicios cloud que no están disponibles en todos los entornos. SageMaker existe en AWS pero no en LocalStack. CloudWatch Metrics funciona en AWS pero es limitado en local. Ciertas integraciones como SNS notifications o SES emails solo tienen sentido en staging/producción. En lugar de hardcodear if environment == "aws" por todo tu código, vas a diseñar un sistema de feature flags que habilita o deshabilita capacidades limpiamente, con fallbacks definidos para cuando una feature no está disponible.

Contexto: En la cápsula 04, tu ClientFactory ya lanza un error si pides SageMaker cuando no está habilitado. Eso es un inicio. Pero en un sistema real, no quieres que toda tu app falle porque una feature opcional no está disponible. Quieres que el flujo principal funcione siempre, y las features opcionales se habiliten cuando el entorno las soporte. Feature flags son el mecanismo para lograr eso — y en cloud, son especialmente importantes porque los entornos tienen capacidades diferentes.

El Problema: Capacidades Diferentes por Entorno

Qué está disponible dónde

Servicio/Feature          LocalStack     AWS Staging    AWS Production
────────────────────────  ────────────   ────────────   ──────────────
S3                        ✅ Completo    ✅ Completo    ✅ Completo
Lambda                    ✅ Completo    ✅ Completo    ✅ Completo
SageMaker Endpoints       ❌ No soporta  ✅ Disponible  ✅ Disponible
CloudWatch Metrics        ⚠️ Parcial     ✅ Completo    ✅ Completo
SNS Notifications         ⚠️ Parcial     ✅ Completo    ✅ Completo
SES Email                 ❌ No soporta  ✅ Sandbox     ✅ Completo
Cost Tracking             ❌ Sin sentido  ✅ Disponible  ✅ Disponible
IAM Enforcement           ❌ No enforced  ✅ Enforced    ✅ Strict

El anti-patrón: condicionales por entorno en lógica de negocio

# ❌ Anti-patrón — tu código de negocio está lleno de condicionales de entorno
import os

def process_document(document: dict) -> dict:
    result = run_inference(document)

    if os.environ.get("ENVIRONMENT") == "production":
        send_sns_notification(result)

    if os.environ.get("ENVIRONMENT") in ("staging", "production"):
        sagemaker_result = invoke_sagemaker(document)
        result["sagemaker_enrichment"] = sagemaker_result

    if os.environ.get("ENVIRONMENT") != "local":
        publish_cloudwatch_metric("documents_processed", 1)

    return result

Problemas:

  • ❌ Lógica de negocio contaminada con decisiones de infraestructura
  • ❌ Cada nuevo developer debe entender todos los condicionales
  • ❌ Agregar un entorno nuevo (QA) requiere revisar cada if
  • ❌ Imposible testear el flujo de producción en local

El Patrón: Feature Flags Declarativos

Diseño del sistema de feature flags

"""services/feature_flags.py — Sistema de feature flags para cloud."""

from dataclasses import dataclass, field
from typing import Any
from config.settings import Settings


@dataclass
class FeatureFlag:
    """Define una feature con su estado y metadata."""
    name: str
    enabled: bool
    description: str
    fallback: Any = None
    requires_service: str | None = None


class FeatureFlags:
    """Gestiona feature flags basados en la configuración del entorno.

    La lógica de negocio pregunta "¿está habilitada esta feature?"
    en lugar de "¿estoy en AWS?"
    """

    def __init__(self, settings: Settings):
        self.settings = settings
        self._flags: dict[str, FeatureFlag] = {}
        self._register_defaults()

    def _register_defaults(self):
        """Registra las feature flags del sistema."""
        self.register(FeatureFlag(
            name="sagemaker_enrichment",
            enabled=self.settings.feature_sagemaker_enabled,
            description="Enriquece resultados con SageMaker model inference",
            fallback={"enriched": False, "reason": "SageMaker not available"},
            requires_service="sagemaker-runtime",
        ))

        self.register(FeatureFlag(
            name="cloudwatch_metrics",
            enabled=self.settings.is_aws,
            description="Publica métricas a CloudWatch",
            fallback=None,
        ))

        self.register(FeatureFlag(
            name="sns_notifications",
            enabled=self.settings.is_aws,
            description="Envía notificaciones via SNS",
            fallback=None,
        ))

        self.register(FeatureFlag(
            name="cost_tracking",
            enabled=self.settings.feature_cost_tracking,
            description="Registra costos estimados por operación",
            fallback=None,
        ))

        self.register(FeatureFlag(
            name="advanced_logging",
            enabled=self.settings.feature_advanced_logging,
            description="Logging detallado con request/response completo",
            fallback=None,
        ))

        self.register(FeatureFlag(
            name="s3_versioning",
            enabled=True,
            description="Versionado de objetos en S3 (funciona en ambos entornos)",
            fallback=None,
        ))

    def register(self, flag: FeatureFlag):
        """Registra una feature flag."""
        self._flags[flag.name] = flag

    def is_enabled(self, name: str) -> bool:
        """Verifica si una feature está habilitada."""
        flag = self._flags.get(name)
        if flag is None:
            return False
        return flag.enabled

    def get_fallback(self, name: str) -> Any:
        """Retorna el valor de fallback para una feature deshabilitada."""
        flag = self._flags.get(name)
        if flag is None:
            return None
        return flag.fallback

    def execute_if_enabled(self, name: str, func, *args, **kwargs) -> Any:
        """Ejecuta una función solo si la feature está habilitada.

        Si está deshabilitada, retorna el fallback.
        Si está habilitada pero falla, retorna el fallback.
        """
        if not self.is_enabled(name):
            return self.get_fallback(name)

        try:
            return func(*args, **kwargs)
        except Exception as e:
            flag = self._flags.get(name)
            if flag:
                import logging
                logging.getLogger(__name__).warning(
                    f"Feature '{name}' habilitada pero falló: {e}. "
                    f"Usando fallback."
                )
            return self.get_fallback(name)

    def status(self) -> dict:
        """Retorna el estado de todas las feature flags."""
        return {
            name: {
                "enabled": flag.enabled,
                "description": flag.description,
                "has_fallback": flag.fallback is not None,
            }
            for name, flag in sorted(self._flags.items())
        }

    def report(self):
        """Imprime un reporte de feature flags."""
        print(f"\nFeature Flags ({self.settings.environment}):")
        for name, flag in sorted(self._flags.items()):
            icon = "✅" if flag.enabled else "❌"
            fallback = " (fallback definido)" if flag.fallback is not None else ""
            print(f"  {icon} {name}: {flag.description}{fallback}")

Uso en Lógica de Negocio

DocumentProcessor con feature flags

"""services/document_processor.py — Processor con feature flags integrados."""

import json
import logging
from datetime import datetime
from typing import Any

logger = logging.getLogger(__name__)


class DocumentProcessor:
    """Procesa documentos con features opcionales controladas por flags."""

    def __init__(
        self,
        s3_client: Any,
        bucket: str,
        feature_flags,
        sagemaker_client: Any = None,
        cloudwatch_client: Any = None,
    ):
        self.s3 = s3_client
        self.bucket = bucket
        self.flags = feature_flags
        self.sagemaker = sagemaker_client
        self.cloudwatch = cloudwatch_client

    def process(self, prompt_name: str, prompt_version: str, document: dict) -> dict:
        """Procesa un documento con todas las features habilitadas."""
        result = self._core_processing(prompt_name, prompt_version, document)

        enrichment = self.flags.execute_if_enabled(
            "sagemaker_enrichment",
            self._enrich_with_sagemaker,
            document,
        )
        if enrichment:
            result["sagemaker_enrichment"] = enrichment

        self.flags.execute_if_enabled(
            "cloudwatch_metrics",
            self._publish_metric,
            "documents_processed",
            1,
        )

        self.flags.execute_if_enabled(
            "cost_tracking",
            self._track_cost,
            result,
        )

        if self.flags.is_enabled("advanced_logging"):
            logger.info(f"Full result: {json.dumps(result, default=str)}")
        else:
            logger.info(f"Processed: {result.get('document_key', 'unknown')}")

        return result

    def _core_processing(
        self, prompt_name: str, prompt_version: str, document: dict
    ) -> dict:
        """Procesamiento core — siempre se ejecuta, sin feature flags."""
        key = f"prompts/{prompt_name}/{prompt_version}/system.txt"
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        template = response["Body"].read().decode("utf-8")

        doc_key = f"documents/inbox/{document.get('id', 'unknown')}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=doc_key,
            Body=json.dumps(document, ensure_ascii=False).encode("utf-8"),
        )

        return {
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "document_key": doc_key,
            "processed": True,
            "template_preview": template[:100],
        }

    def _enrich_with_sagemaker(self, document: dict) -> dict:
        """Enriquece con SageMaker (solo AWS)."""
        if not self.sagemaker:
            return {"enriched": False, "reason": "SageMaker client not provided"}

        response = self.sagemaker.invoke_endpoint(
            EndpointName="document-enrichment",
            ContentType="application/json",
            Body=json.dumps(document).encode("utf-8"),
        )
        payload = json.loads(response["Body"].read().decode("utf-8"))
        return {"enriched": True, "data": payload}

    def _publish_metric(self, metric_name: str, value: float):
        """Publica métrica a CloudWatch (solo AWS)."""
        if not self.cloudwatch:
            return

        self.cloudwatch.put_metric_data(
            Namespace="AIService",
            MetricData=[{
                "MetricName": metric_name,
                "Value": value,
                "Unit": "Count",
            }],
        )

    def _track_cost(self, result: dict):
        """Registra costo estimado (solo cuando cost tracking está habilitado)."""
        estimated_cost = 0.0001
        now = datetime.utcnow()
        cost_key = f"costs/{now.strftime('%Y/%m/%d')}/estimate.json"

        try:
            existing = self.s3.get_object(Bucket=self.bucket, Key=cost_key)
            costs = json.loads(existing["Body"].read().decode("utf-8"))
        except Exception:
            costs = {"date": now.strftime("%Y-%m-%d"), "total": 0, "operations": 0}

        costs["total"] += estimated_cost
        costs["operations"] += 1

        self.s3.put_object(
            Bucket=self.bucket,
            Key=cost_key,
            Body=json.dumps(costs).encode("utf-8"),
        )

Wiring con feature flags

"""app.py — Wiring con feature flags."""

from config.loader import get_settings
from clients.factory import ClientFactory
from services.feature_flags import FeatureFlags
from services.document_processor import DocumentProcessor


def create_app():
    settings = get_settings()
    factory = ClientFactory(settings)
    flags = FeatureFlags(settings)

    flags.report()

    sagemaker_client = None
    if flags.is_enabled("sagemaker_enrichment"):
        try:
            sagemaker_client = factory.get_client("sagemaker-runtime")
        except Exception as e:
            import logging
            logging.warning(f"SageMaker client no disponible: {e}")

    cloudwatch_client = None
    if flags.is_enabled("cloudwatch_metrics"):
        cloudwatch_client = factory.get_client("cloudwatch")

    processor = DocumentProcessor(
        s3_client=factory.s3,
        bucket=settings.s3_bucket,
        feature_flags=flags,
        sagemaker_client=sagemaker_client,
        cloudwatch_client=cloudwatch_client,
    )

    return {
        "settings": settings,
        "factory": factory,
        "flags": flags,
        "processor": processor,
    }

Feature Flags Dinámicos: Cambiar sin Re-deploy

Flags desde S3 (runtime update)

"""services/dynamic_flags.py — Feature flags que se actualizan en runtime."""

import json
import time
import logging
from typing import Any

logger = logging.getLogger(__name__)


class DynamicFeatureFlags:
    """Feature flags que se recargan desde S3 periódicamente.

    Permite habilitar/deshabilitar features sin re-deploy.
    """

    def __init__(
        self,
        s3_client: Any,
        bucket: str,
        config_key: str = "config/feature-flags.json",
        refresh_interval: int = 60,
    ):
        self.s3 = s3_client
        self.bucket = bucket
        self.config_key = config_key
        self.refresh_interval = refresh_interval
        self._flags: dict[str, bool] = {}
        self._last_refresh: float = 0
        self._defaults: dict[str, bool] = {}

    def set_defaults(self, defaults: dict[str, bool]):
        """Establece defaults para cuando S3 no está disponible."""
        self._defaults = defaults
        self._flags = {**defaults}

    def refresh(self) -> bool:
        """Recarga flags desde S3 si ha pasado el intervalo."""
        now = time.time()
        if now - self._last_refresh < self.refresh_interval:
            return False

        try:
            response = self.s3.get_object(
                Bucket=self.bucket, Key=self.config_key
            )
            content = response["Body"].read().decode("utf-8")
            remote_flags = json.loads(content)

            self._flags = {**self._defaults, **remote_flags}
            self._last_refresh = now
            logger.info(f"Feature flags recargados desde S3: {self._flags}")
            return True

        except Exception as e:
            logger.warning(
                f"No se pudieron recargar feature flags desde S3: {e}. "
                f"Usando valores actuales."
            )
            self._last_refresh = now
            return False

    def is_enabled(self, name: str) -> bool:
        self.refresh()
        return self._flags.get(name, False)

    def upload_flags(self, flags: dict[str, bool]):
        """Sube nuevos flags a S3 (para administración)."""
        self.s3.put_object(
            Bucket=self.bucket,
            Key=self.config_key,
            Body=json.dumps(flags, indent=2).encode("utf-8"),
            ContentType="application/json",
        )
        logger.info(f"Feature flags actualizados en S3: {flags}")
        self._last_refresh = 0  # Forzar recarga en próximo is_enabled

Uso de flags dinámicos

from config.loader import get_settings
from clients.factory import ClientFactory
from services.dynamic_flags import DynamicFeatureFlags

settings = get_settings()
factory = ClientFactory(settings)

dynamic_flags = DynamicFeatureFlags(
    s3_client=factory.s3,
    bucket=settings.s3_bucket,
    refresh_interval=30,
)

dynamic_flags.set_defaults({
    "sagemaker_enrichment": settings.feature_sagemaker_enabled,
    "cloudwatch_metrics": settings.is_aws,
    "new_model_v2": False,  # Feature nueva, deshabilitada por default
})

# Subir flags iniciales a S3
dynamic_flags.upload_flags({
    "sagemaker_enrichment": True,
    "cloudwatch_metrics": True,
    "new_model_v2": False,
})

# En runtime, verificar feature
if dynamic_flags.is_enabled("new_model_v2"):
    result = invoke_new_model(document)
else:
    result = invoke_current_model(document)

# Para habilitar sin re-deploy:
# dynamic_flags.upload_flags({"new_model_v2": True})
# → En 30 segundos, todos los workers lo recogen

Health Endpoint con Feature Flags

/health que reporta estado de features

"""handler.py — Health endpoint con feature flags."""

import json


def health_handler(container) -> dict:
    """Health check que incluye estado de feature flags."""
    flags_status = container.flags.status()
    client_health = container.factory.health_check()

    features_summary = {}
    for name, info in flags_status.items():
        if info["enabled"]:
            service = container.flags._flags.get(name)
            if service and service.requires_service:
                svc = service.requires_service
                features_summary[name] = {
                    "enabled": True,
                    "service_health": client_health.get(svc, "unknown"),
                }
            else:
                features_summary[name] = {"enabled": True}
        else:
            features_summary[name] = {
                "enabled": False,
                "has_fallback": info["has_fallback"],
            }

    return {
        "statusCode": 200,
        "body": json.dumps({
            "status": "healthy",
            "environment": container.settings.environment,
            "features": features_summary,
            "services": client_health,
        }),
    }

Ejemplo de response en LocalStack:

{
  "status": "healthy",
  "environment": "local",
  "features": {
    "sagemaker_enrichment": {"enabled": false, "has_fallback": true},
    "cloudwatch_metrics": {"enabled": false, "has_fallback": false},
    "cost_tracking": {"enabled": false, "has_fallback": false},
    "advanced_logging": {"enabled": false, "has_fallback": false},
    "s3_versioning": {"enabled": true}
  },
  "services": {
    "s3": "healthy"
  }
}

Ejemplo en AWS staging:

{
  "status": "healthy",
  "environment": "staging",
  "features": {
    "sagemaker_enrichment": {"enabled": true, "service_health": "healthy"},
    "cloudwatch_metrics": {"enabled": true},
    "cost_tracking": {"enabled": true},
    "advanced_logging": {"enabled": true},
    "s3_versioning": {"enabled": true}
  },
  "services": {
    "s3": "healthy",
    "sagemaker-runtime": "healthy",
    "cloudwatch": "healthy"
  }
}

Troubleshooting

Problema 1: Feature flag habilitado pero el servicio no responde

La feature está enabled=True pero SageMaker no tiene endpoint desplegado.

# execute_if_enabled ya maneja esto — si la función lanza excepción,
# retorna el fallback. Pero necesitas verificar en logs:

# Logger output:
# WARNING: Feature 'sagemaker_enrichment' habilitada pero falló:
# EndpointNotFound. Usando fallback.

# Solución: verificar que el endpoint existe antes de habilitar
# o usar el health_check del factory para detección automática

Problema 2: Flags dinámicos no se actualizan

El refresh_interval no ha pasado, o S3 no es accesible.

# Verificar:
print(f"Last refresh: {dynamic_flags._last_refresh}")
print(f"Interval: {dynamic_flags.refresh_interval}")
print(f"Current flags: {dynamic_flags._flags}")

# Forzar refresh:
dynamic_flags._last_refresh = 0
dynamic_flags.refresh()

Problema 3: Feature flag en .env no coincide con el código

El .env tiene FEATURE_SAGEMAKER_ENABLED=true pero el Settings tiene otro nombre.

# Pydantic Settings mapea FEATURE_SAGEMAKER_ENABLED → feature_sagemaker_enabled
# La convención es: VARIABLE_DE_ENTORNO en UPPER_SNAKE_CASE
# se mapea a campo Python en lower_snake_case

# Si no coinciden, verifica:
settings = Settings()
print(f"feature_sagemaker_enabled: {settings.feature_sagemaker_enabled}")

Problema 4: Tests ignoran feature flags

Tests unitarios con mock no pasan por execute_if_enabled.

# Solución: testear con FeatureFlags mockeado
from unittest.mock import MagicMock

mock_flags = MagicMock()
mock_flags.is_enabled.return_value = True
mock_flags.execute_if_enabled.side_effect = lambda name, func, *a, **kw: func(*a, **kw)

# O crear FeatureFlags con settings de test:
test_settings = Settings(feature_sagemaker_enabled=True)
flags = FeatureFlags(test_settings)
assert flags.is_enabled("sagemaker_enrichment")

Ejercicios Prácticos

Ejercicio 1: Feature flag con porcentaje de rollout

Implementa un feature flag que se habilite para un porcentaje de requests (ej: 10% de las invocaciones usan el modelo nuevo, 90% usan el actual).

Ver solución 
import random
from dataclasses import dataclass


@dataclass
class GradualFeatureFlag:
    """Feature flag con rollout gradual por porcentaje."""
    name: str
    description: str
    rollout_percentage: float  # 0.0 a 1.0
    enabled: bool = True

    def should_activate(self, request_id: str | None = None) -> bool:
        """Decide si activar la feature para este request.

        Si se provee request_id, el resultado es determinístico
        (mismo request_id → mismo resultado).
        """
        if not self.enabled:
            return False

        if self.rollout_percentage >= 1.0:
            return True

        if self.rollout_percentage <= 0.0:
            return False

        if request_id:
            hash_value = hash(request_id) % 100
            return hash_value < (self.rollout_percentage * 100)

        return random.random() < self.rollout_percentage


class GradualFlags:
    def __init__(self):
        self._flags: dict[str, GradualFeatureFlag] = {}

    def register(self, flag: GradualFeatureFlag):
        self._flags[flag.name] = flag

    def should_activate(self, name: str, request_id: str | None = None) -> bool:
        flag = self._flags.get(name)
        if not flag:
            return False
        return flag.should_activate(request_id)

    def report(self):
        for name, flag in self._flags.items():
            pct = flag.rollout_percentage * 100
            status = "✅" if flag.enabled else "❌"
            print(f"  {status} {name}: {pct:.0f}% rollout — {flag.description}")


# Uso
flags = GradualFlags()
flags.register(GradualFeatureFlag(
    name="new_model_v2",
    description="Nuevo modelo de clasificación",
    rollout_percentage=0.1,
))

flags.report()

# Simular 100 requests
activations = sum(
    flags.should_activate("new_model_v2", f"req-{i}")
    for i in range(100)
)
print(f"\nActivaciones en 100 requests: {activations} (~10 esperados)")

Ejercicio 2: Feature flag con dependencias

Crea un sistema donde una feature depende de otra (ej: "cost_tracking" requiere "cloudwatch_metrics"). Si la dependencia no está habilitada, la feature dependiente tampoco se habilita.

Ver solución 
from dataclasses import dataclass, field


@dataclass
class DependentFeatureFlag:
    name: str
    enabled: bool
    description: str
    depends_on: list[str] = field(default_factory=list)


class DependencyAwareFlags:
    def __init__(self):
        self._flags: dict[str, DependentFeatureFlag] = {}

    def register(self, flag: DependentFeatureFlag):
        self._flags[flag.name] = flag

    def is_enabled(self, name: str, visited: set | None = None) -> bool:
        """Verifica si una feature está habilitada, considerando dependencias."""
        if visited is None:
            visited = set()

        if name in visited:
            return False  # Circular dependency protection
        visited.add(name)

        flag = self._flags.get(name)
        if not flag:
            return False

        if not flag.enabled:
            return False

        for dep in flag.depends_on:
            if not self.is_enabled(dep, visited):
                return False

        return True

    def status(self) -> dict:
        result = {}
        for name, flag in self._flags.items():
            effective = self.is_enabled(name)
            blocked_by = []
            if flag.enabled and not effective:
                for dep in flag.depends_on:
                    if not self.is_enabled(dep):
                        blocked_by.append(dep)
            result[name] = {
                "configured": flag.enabled,
                "effective": effective,
                "blocked_by": blocked_by if blocked_by else None,
            }
        return result


# Uso
flags = DependencyAwareFlags()
flags.register(DependentFeatureFlag(
    name="cloudwatch_metrics",
    enabled=True,
    description="Publish metrics to CloudWatch",
))
flags.register(DependentFeatureFlag(
    name="cost_tracking",
    enabled=True,
    description="Track estimated costs",
    depends_on=["cloudwatch_metrics"],
))
flags.register(DependentFeatureFlag(
    name="cost_alerts",
    enabled=True,
    description="Alert on cost thresholds",
    depends_on=["cost_tracking", "sns_notifications"],
))
flags.register(DependentFeatureFlag(
    name="sns_notifications",
    enabled=False,  # Deshabilitado
    description="Send SNS notifications",
))

status = flags.status()
for name, info in status.items():
    icon = "✅" if info["effective"] else "❌"
    blocked = f" (blocked by: {info['blocked_by']})" if info["blocked_by"] else ""
    print(f"  {icon} {name}: configured={info['configured']}, effective={info['effective']}{blocked}")

Ejercicio 3: Feature flags audit log

Implementa un sistema que registra cada vez que una feature flag se evalúa, creando un audit log que se puede analizar para entender el uso de features por entorno.

Ver solución 
import json
from datetime import datetime
from collections import defaultdict


class AuditedFeatureFlags:
    """Feature flags con audit log de evaluaciones."""

    def __init__(self, flags: dict[str, bool], environment: str):
        self._flags = flags
        self._environment = environment
        self._audit_log: list[dict] = []
        self._stats: dict[str, dict] = defaultdict(
            lambda: {"checked": 0, "enabled": 0, "disabled": 0}
        )

    def is_enabled(self, name: str, context: dict | None = None) -> bool:
        enabled = self._flags.get(name, False)

        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "environment": self._environment,
            "flag": name,
            "result": enabled,
            "context": context,
        }
        self._audit_log.append(entry)

        self._stats[name]["checked"] += 1
        if enabled:
            self._stats[name]["enabled"] += 1
        else:
            self._stats[name]["disabled"] += 1

        return enabled

    def get_audit_log(self, flag_name: str | None = None) -> list[dict]:
        if flag_name:
            return [e for e in self._audit_log if e["flag"] == flag_name]
        return self._audit_log

    def get_stats(self) -> dict:
        return dict(self._stats)

    def export_audit(self, filepath: str):
        with open(filepath, "w") as f:
            json.dump({
                "environment": self._environment,
                "exported_at": datetime.utcnow().isoformat(),
                "total_evaluations": len(self._audit_log),
                "stats": self.get_stats(),
                "log": self._audit_log,
            }, f, indent=2)
        print(f"Audit log exportado: {filepath} ({len(self._audit_log)} entries)")


# Uso
flags = AuditedFeatureFlags(
    flags={"sagemaker_enrichment": False, "cost_tracking": True},
    environment="local",
)

for i in range(10):
    flags.is_enabled("sagemaker_enrichment", context={"request_id": f"req-{i}"})
    flags.is_enabled("cost_tracking", context={"request_id": f"req-{i}"})

print("Stats:")
for flag, stats in flags.get_stats().items():
    print(f"  {flag}: {stats}")

flags.export_audit("feature-flags-audit.json")

Ejercicio 4: Migration checklist basada en feature flags

Crea una función que compare las feature flags entre dos entornos y genere un checklist de migración: qué features se van a habilitar, cuáles se van a deshabilitar, y qué servicios necesitan estar disponibles.

Ver solución 
from config.settings import Settings, EnvironmentName
from services.feature_flags import FeatureFlags


def generate_migration_checklist(
    source_env: str,
    target_env: str,
) -> dict:
    """Genera checklist de migración basada en feature flags."""
    source_settings = Settings(environment=source_env)
    target_settings = Settings(environment=target_env)

    source_flags = FeatureFlags(source_settings)
    target_flags = FeatureFlags(target_settings)

    checklist = {
        "migration": f"{source_env} → {target_env}",
        "newly_enabled": [],
        "newly_disabled": [],
        "unchanged": [],
        "services_required": [],
        "warnings": [],
    }

    all_flag_names = set(
        list(source_flags._flags.keys()) + list(target_flags._flags.keys())
    )

    for name in sorted(all_flag_names):
        source_enabled = source_flags.is_enabled(name)
        target_enabled = target_flags.is_enabled(name)

        if source_enabled == target_enabled:
            checklist["unchanged"].append(name)
        elif target_enabled and not source_enabled:
            checklist["newly_enabled"].append(name)
            flag = target_flags._flags.get(name)
            if flag and flag.requires_service:
                checklist["services_required"].append({
                    "feature": name,
                    "service": flag.requires_service,
                })
        else:
            checklist["newly_disabled"].append(name)
            checklist["warnings"].append(
                f"Feature '{name}' se deshabilita en {target_env}. "
                f"Verificar que el fallback es aceptable."
            )

    print(f"\n{'='*60}")
    print(f"MIGRATION CHECKLIST: {source_env} → {target_env}")
    print(f"{'='*60}")

    if checklist["newly_enabled"]:
        print(f"\n🟢 Features que se HABILITAN:")
        for name in checklist["newly_enabled"]:
            print(f"   ✅ {name}")

    if checklist["newly_disabled"]:
        print(f"\n🔴 Features que se DESHABILITAN:")
        for name in checklist["newly_disabled"]:
            print(f"   ❌ {name}")

    if checklist["services_required"]:
        print(f"\n🔧 Servicios REQUERIDOS en {target_env}:")
        for req in checklist["services_required"]:
            print(f"   → {req['service']} (para {req['feature']})")

    if checklist["warnings"]:
        print(f"\n⚠️ WARNINGS:")
        for w in checklist["warnings"]:
            print(f"   {w}")

    return checklist


generate_migration_checklist("local", "staging")

Resumen

  • Feature flags desacoplan capacidades de entornos. Tu código pregunta "¿está habilitada esta feature?" en lugar de "¿estoy en AWS?" La diferencia es que agregar un entorno nuevo no requiere tocar lógica de negocio.
  • execute_if_enabled es el método central. Ejecuta la función si está habilitada, retorna el fallback si no. Maneja excepciones gracefully.
  • Flags estáticos (config) vs dinámicos (S3). Config-based para flags que no cambian en runtime. S3-based para cambiar sin re-deploy.
  • El health endpoint reporta feature flags. Cada entorno muestra qué features están activas y cuál es la salud de los servicios que requieren.
  • Feature flags facilitan la migración. Puedes migrar en pasos: primero migra sin SageMaker (FEATURE_SAGEMAKER_ENABLED=false), verifica, luego habilita SageMaker.
  • En la siguiente cápsula, extenderás esto con graceful degradation — qué pasa cuando un servicio que está habilitado falla en runtime.

Recursos Adicionales

  1. Martin Fowler — Feature Toggles — El artículo definitivo sobre feature flags
  2. LaunchDarkly — Feature Flag Best Practices — Best practices de feature flags
  3. AWS AppConfig — Feature flags como servicio en AWS
  4. 12 Factor App — Config — Configuración externalizada
  5. Feature Flags in Python — Implementaciones de feature flags en Python
  6. Gradual Rollout Strategies — Estrategias de rollout gradual
← Anterior5. Testing Multi-EntornoSiguiente →7. Graceful Degradation