Module 6: Cloud Migration Patterns
6. Cloud Feature Flags
Overview
In this capsule you'll implement feature flags to handle cloud services that aren't available in every environment. SageMaker exists on AWS but not on LocalStack. CloudWatch Metrics works on AWS but is limited locally. Certain integrations like SNS notifications or SES emails only make sense in staging/production. Instead of hardcoding if environment == "aws" throughout your code, you'll design a feature flags system that enables or disables capabilities cleanly, with defined fallbacks for when a feature isn't available.
Context: In capsule 04, your ClientFactory already raises an error if you ask for SageMaker when it isn't enabled. That's a start. But in a real system, you don't want your entire app to fail because an optional feature isn't available. You want the main flow to always work, and the optional features to be enabled when the environment supports them. Feature flags are the mechanism to achieve that — and in the cloud, they're especially important because environments have different capabilities.
The Problem: Different Capabilities per Environment
What's available where
Service/Feature LocalStack AWS Staging AWS Production
──────────────────────── ──────────── ──────────── ──────────────
S3 ✅ Complete ✅ Complete ✅ Complete
Lambda ✅ Complete ✅ Complete ✅ Complete
SageMaker Endpoints ❌ Unsupported ✅ Available ✅ Available
CloudWatch Metrics ⚠️ Partial ✅ Complete ✅ Complete
SNS Notifications ⚠️ Partial ✅ Complete ✅ Complete
SES Email ❌ Unsupported ✅ Sandbox ✅ Complete
Cost Tracking ❌ No purpose ✅ Available ✅ Available
IAM Enforcement ❌ No enforced ✅ Enforced ✅ Strict
The anti-pattern: per-environment conditionals in business logic
# ❌ Anti-pattern — your business code is full of environment conditionals
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 resultProblems:
- ❌ Business logic contaminated with infrastructure decisions
- ❌ Every new developer must understand all the conditionals
- ❌ Adding a new environment (QA) requires reviewing every
if - ❌ Impossible to test the production flow locally
The Pattern: Declarative Feature Flags
Design of the feature flags system
"""services/feature_flags.py — Feature flags system for the cloud."""
from dataclasses import dataclass, field
from typing import Any
from config.settings import Settings
@dataclass
class FeatureFlag:
"""Defines a feature with its state and metadata."""
name: str
enabled: bool
description: str
fallback: Any = None
requires_service: str | None = None
class FeatureFlags:
"""Manages feature flags based on the environment's configuration.
The business logic asks "is this feature enabled?"
instead of "am I on AWS?"
"""
def __init__(self, settings: Settings):
self.settings = settings
self._flags: dict[str, FeatureFlag] = {}
self._register_defaults()
def _register_defaults(self):
"""Registers the system's feature flags."""
self.register(FeatureFlag(
name="sagemaker_enrichment",
enabled=self.settings.feature_sagemaker_enabled,
description="Enriches results with 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="Publishes metrics to CloudWatch",
fallback=None,
))
self.register(FeatureFlag(
name="sns_notifications",
enabled=self.settings.is_aws,
description="Sends notifications via SNS",
fallback=None,
))
self.register(FeatureFlag(
name="cost_tracking",
enabled=self.settings.feature_cost_tracking,
description="Records estimated costs per operation",
fallback=None,
))
self.register(FeatureFlag(
name="advanced_logging",
enabled=self.settings.feature_advanced_logging,
description="Detailed logging with the full request/response",
fallback=None,
))
self.register(FeatureFlag(
name="s3_versioning",
enabled=True,
description="Object versioning in S3 (works in both environments)",
fallback=None,
))
def register(self, flag: FeatureFlag):
"""Registers a feature flag."""
self._flags[flag.name] = flag
def is_enabled(self, name: str) -> bool:
"""Checks whether a feature is enabled."""
flag = self._flags.get(name)
if flag is None:
return False
return flag.enabled
def get_fallback(self, name: str) -> Any:
"""Returns the fallback value for a disabled feature."""
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:
"""Runs a function only if the feature is enabled.
If it's disabled, returns the fallback.
If it's enabled but fails, returns the 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}' enabled but failed: {e}. "
f"Using fallback."
)
return self.get_fallback(name)
def status(self) -> dict:
"""Returns the state of all 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):
"""Prints a feature flags report."""
print(f"\nFeature Flags ({self.settings.environment}):")
for name, flag in sorted(self._flags.items()):
icon = "✅" if flag.enabled else "❌"
fallback = " (fallback defined)" if flag.fallback is not None else ""
print(f" {icon} {name}: {flag.description}{fallback}")Use in Business Logic
DocumentProcessor with feature flags
"""services/document_processor.py — Processor with integrated feature flags."""
import json
import logging
from datetime import datetime
from typing import Any
logger = logging.getLogger(__name__)
class DocumentProcessor:
"""Processes documents with optional features controlled by 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:
"""Processes a document with all the enabled features."""
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:
"""Core processing — always runs, no 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:
"""Enriches with SageMaker (AWS only)."""
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):
"""Publishes a metric to CloudWatch (AWS only)."""
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):
"""Records the estimated cost (only when cost tracking is enabled)."""
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 with feature flags
"""app.py — Wiring with 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 not available: {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,
}Dynamic Feature Flags: Change Without Re-deploying
Flags from S3 (runtime update)
"""services/dynamic_flags.py — Feature flags that update at runtime."""
import json
import time
import logging
from typing import Any
logger = logging.getLogger(__name__)
class DynamicFeatureFlags:
"""Feature flags that reload from S3 periodically.
Allows enabling/disabling features without re-deploying.
"""
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]):
"""Sets defaults for when S3 isn't available."""
self._defaults = defaults
self._flags = {**defaults}
def refresh(self) -> bool:
"""Reloads flags from S3 if the interval has passed."""
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 reloaded from S3: {self._flags}")
return True
except Exception as e:
logger.warning(
f"Could not reload feature flags from S3: {e}. "
f"Using current values."
)
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]):
"""Uploads new flags to S3 (for administration)."""
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 updated in S3: {flags}")
self._last_refresh = 0 # Force a reload on the next is_enabledUsing dynamic flags
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, # New feature, disabled by default
})
# Upload the initial flags to S3
dynamic_flags.upload_flags({
"sagemaker_enrichment": True,
"cloudwatch_metrics": True,
"new_model_v2": False,
})
# At runtime, check the feature
if dynamic_flags.is_enabled("new_model_v2"):
result = invoke_new_model(document)
else:
result = invoke_current_model(document)
# To enable without re-deploying:
# dynamic_flags.upload_flags({"new_model_v2": True})
# → Within 30 seconds, all the workers pick it upHealth Endpoint with Feature Flags
/health that reports the state of features
"""handler.py — Health endpoint with feature flags."""
import json
def health_handler(container) -> dict:
"""Health check that includes the state of 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,
}),
}Example response on 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"
}
}Example on 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
Problem 1: Feature flag enabled but the service doesn't respond
The feature is enabled=True but SageMaker doesn't have an endpoint deployed.
# execute_if_enabled already handles this — if the function raises an exception,
# it returns the fallback. But you need to check the logs:
# Logger output:
# WARNING: Feature 'sagemaker_enrichment' enabled but failed:
# EndpointNotFound. Using fallback.
# Solution: verify that the endpoint exists before enabling it
# or use the factory's health_check for automatic detectionProblem 2: Dynamic flags don't update
The refresh_interval hasn't passed, or S3 isn't accessible.
# Check:
print(f"Last refresh: {dynamic_flags._last_refresh}")
print(f"Interval: {dynamic_flags.refresh_interval}")
print(f"Current flags: {dynamic_flags._flags}")
# Force a refresh:
dynamic_flags._last_refresh = 0
dynamic_flags.refresh()Problem 3: Feature flag in .env doesn't match the code
The .env has FEATURE_SAGEMAKER_ENABLED=true but Settings has another name.
# Pydantic Settings maps FEATURE_SAGEMAKER_ENABLED → feature_sagemaker_enabled
# The convention is: ENVIRONMENT_VARIABLE in UPPER_SNAKE_CASE
# maps to a Python field in lower_snake_case
# If they don't match, check:
settings = Settings()
print(f"feature_sagemaker_enabled: {settings.feature_sagemaker_enabled}")Problem 4: Tests ignore feature flags
Unit tests with mocks don't go through execute_if_enabled.
# Solution: test with a mocked FeatureFlags
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)
# Or create FeatureFlags with test settings:
test_settings = Settings(feature_sagemaker_enabled=True)
flags = FeatureFlags(test_settings)
assert flags.is_enabled("sagemaker_enrichment")Practical Exercises
Exercise 1: Feature flag with rollout percentage
Implement a feature flag that's enabled for a percentage of requests (e.g., 10% of invocations use the new model, 90% use the current one).
See solution
import random
from dataclasses import dataclass
@dataclass
class GradualFeatureFlag:
"""Feature flag with a gradual rollout by percentage."""
name: str
description: str
rollout_percentage: float # 0.0 to 1.0
enabled: bool = True
def should_activate(self, request_id: str | None = None) -> bool:
"""Decides whether to activate the feature for this request.
If request_id is provided, the result is deterministic
(same request_id → same result).
"""
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}")
# Usage
flags = GradualFlags()
flags.register(GradualFeatureFlag(
name="new_model_v2",
description="New classification model",
rollout_percentage=0.1,
))
flags.report()
# Simulate 100 requests
activations = sum(
flags.should_activate("new_model_v2", f"req-{i}")
for i in range(100)
)
print(f"\nActivations in 100 requests: {activations} (~10 expected)")Exercise 2: Feature flag with dependencies
Create a system where one feature depends on another (e.g., "cost_tracking" requires "cloudwatch_metrics"). If the dependency isn't enabled, the dependent feature isn't enabled either.
See solution
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:
"""Checks whether a feature is enabled, considering dependencies."""
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
# Usage
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, # Disabled
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}")Exercise 3: Feature flags audit log
Implement a system that records every time a feature flag is evaluated, creating an audit log that can be analyzed to understand feature usage per environment.
See solution
import json
from datetime import datetime
from collections import defaultdict
class AuditedFeatureFlags:
"""Feature flags with an audit log of evaluations."""
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 exported: {filepath} ({len(self._audit_log)} entries)")
# Usage
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")Exercise 4: Migration checklist based on feature flags
Create a function that compares the feature flags between two environments and generates a migration checklist: which features will be enabled, which will be disabled, and which services need to be available.
See solution
from config.settings import Settings, EnvironmentName
from services.feature_flags import FeatureFlags
def generate_migration_checklist(
source_env: str,
target_env: str,
) -> dict:
"""Generates a migration checklist based on 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}' is disabled in {target_env}. "
f"Verify that the fallback is acceptable."
)
print(f"\n{'='*60}")
print(f"MIGRATION CHECKLIST: {source_env} → {target_env}")
print(f"{'='*60}")
if checklist["newly_enabled"]:
print(f"\n🟢 Features being ENABLED:")
for name in checklist["newly_enabled"]:
print(f" ✅ {name}")
if checklist["newly_disabled"]:
print(f"\n🔴 Features being DISABLED:")
for name in checklist["newly_disabled"]:
print(f" ❌ {name}")
if checklist["services_required"]:
print(f"\n🔧 Services REQUIRED in {target_env}:")
for req in checklist["services_required"]:
print(f" → {req['service']} (for {req['feature']})")
if checklist["warnings"]:
print(f"\n⚠️ WARNINGS:")
for w in checklist["warnings"]:
print(f" {w}")
return checklist
generate_migration_checklist("local", "staging")Summary
- Feature flags decouple capabilities from environments. Your code asks "is this feature enabled?" instead of "am I on AWS?" The difference is that adding a new environment doesn't require touching business logic.
execute_if_enabledis the central method. It runs the function if it's enabled, returns the fallback if not. It handles exceptions gracefully.- Static flags (config) vs dynamic (S3). Config-based for flags that don't change at runtime. S3-based to change without re-deploying.
- The health endpoint reports feature flags. Each environment shows which features are active and the health of the services they require.
- Feature flags make migration easier. You can migrate in steps: first migrate without SageMaker (
FEATURE_SAGEMAKER_ENABLED=false), verify, then enable SageMaker. - In the next capsule, you'll extend this with graceful degradation — what happens when a service that's enabled fails at runtime.
Additional Resources
- Martin Fowler — Feature Toggles — The definitive article on feature flags
- LaunchDarkly — Feature Flag Best Practices — Feature flag best practices
- AWS AppConfig — Feature flags as a service on AWS
- 12 Factor App — Config — Externalized configuration
- Feature Flags in Python — Feature flag implementations in Python
- Gradual Rollout Strategies — Gradual rollout strategies