8. Project: Migration-Ready AI App

Project description

This is the integrative project for Module 6. You will refactor the AI app from the previous modules into a migration-ready application: the same code runs against LocalStack in development and against AWS in staging/production without changing a single line of business logic. The application has environment abstraction, config management with Pydantic Settings, dependency injection for boto3 clients, multi-environment tests, feature flags for SageMaker, graceful degradation with circuit breakers, and a documented migration runbook that another engineer can follow step by step.

Why it matters: This project integrates everything you learned in the module's capsules: environment abstraction (C02), config management (C03), dependency injection (C04), multi-environment testing (C05), feature flags (C06), and graceful degradation (C07). It's the most sophisticated artifact of Phase 2 and the one you'll carry into the Integrative Project (M8). When you finish, you'll have an app that proves you know how to build production software — not prototypes that "work on my machine."

---

Project goal

Produce a functional Migration-Ready AI App that:

1. Runs against LocalStack with ENVIRONMENT=local without changing code
2. Runs against AWS with ENVIRONMENT=staging or ENVIRONMENT=production without changing code
3. Has typed config management with Pydantic Settings and per-environment .env files
4. Uses dependency injection: a factory creates all boto3 clients based on the environment
5. Includes feature flags: SageMaker enabled only on AWS, graceful fallback in local
6. Implements graceful degradation: circuit breakers and fallbacks for S3 and Lambda
7. Has a test suite that passes against LocalStack AND against AWS
8. Includes an operational migration runbook: concrete steps to migrate from local to AWS
9. Has a health endpoint that reports degradation level and available features

---

Module recap

Capsule	Concept	How you use it in the project
02	Environment Abstraction	Environment-agnostic code across the whole project
03	Config Management	Pydantic Settings, .env files, validation
04	Dependency Injection	ClientFactory, ServiceContainer
05	Multi-Environment Testing	conftest.py, markers, universal and per-environment tests
06	Feature Flags	SageMaker flag, execute_if_enabled
07	Graceful Degradation	Circuit breakers, fallbacks, health levels

---

Technical Specifications

Architecture

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: ✅       │
│                                           └──────────────────────────┘
│                    ▲                                   ▲
│                    │                                   │
│                    └───────────┐      ┌────────────────┘
│                                │      │
│                    ┌───────────┴──────┴────────────┐
│                    │        SAME SOURCE CODE        │
│                    │                                │
│                    │  config/settings.py            │
│                    │  clients/factory.py            │
│                    │  services/processor.py         │
│                    │  services/feature_flags.py     │
│                    │  services/health.py            │
│                    │  handler.py                    │
│                    │  tests/conftest.py             │
│                    └────────────────────────────────┘

File structure

migration-ready-app/
├── config/
│   ├── __init__.py
│   ├── settings.py              ← Pydantic Settings with validation
│   ├── loader.py                ← get_settings() with env detection
│   └── validators.py            ← Per-environment validation
├── clients/
│   ├── __init__.py
│   ├── factory.py               ← ClientFactory (boto3 clients)
│   └── resilient.py             ← ResilientClient wrapper
├── services/
│   ├── __init__.py
│   ├── container.py             ← ServiceContainer (DI wiring)
│   ├── document_processor.py    ← Business logic
│   ├── feature_flags.py         ← Feature flags
│   ├── circuit_breaker.py       ← Circuit breaker
│   ├── fallbacks.py             ← Fallback strategies
│   └── health.py                ← Health checker with degradation
├── tests/
│   ├── conftest.py              ← Multi-environment fixtures
│   ├── test_s3_operations.py    ← Universal S3 tests
│   ├── test_processor.py        ← Processor tests
│   ├── test_feature_flags.py    ← Feature flag tests
│   ├── test_degradation.py      ← Graceful degradation tests
│   ├── test_aws_only.py         ← AWS-only tests
│   └── test_migration.py        ← End-to-end migration tests
├── migration/
│   └── RUNBOOK.md               ← Step-by-step migration runbook
├── .env.local                   ← LocalStack config
├── .env.staging                 ← AWS staging config
├── .env.production              ← AWS production config
├── .env.example                 ← Template for new developers
├── .gitignore                   ← Excludes .env files
├── handler.py                   ← Lambda handler (entry point)
├── pytest.ini                   ← pytest configuration
└── requirements.txt             ← Dependencies

Required endpoints

POST /process  → Processes a document with a prompt template
GET  /health   → Status with degradation levels and features

Request/Response format

// POST /process — Request
{
  "prompt_name": "summarizer",
  "prompt_version": "v1",
  "document": {
    "id": "doc-001",
    "title": "Deployment guide",
    "content": "Deploying AI applications requires..."
  }
}

// POST /process — Response (healthy)
{
  "status": "ok",
  "data": {
    "prompt_used": "summarizer/v1",
    "document_stored": "documents/inbox/doc-001.json",
    "processed": true,
    "template_preview": "Summarize the document in 3 points...",
    "sagemaker_enrichment": null,
    "degraded": false,
    "degradation_details": []
  }
}

// POST /process — Response (degraded, slow S3)
{
  "status": "partial",
  "data": {
    "prompt_used": "summarizer/v1",
    "document_stored": null,
    "processed": true,
    "template_preview": "Generate a concise summary...",
    "degraded": true,
    "degradation_details": [
      "Storage: document not persisted (CircuitBreakerError)"
    ]
  },
  "degradation": {
    "level": "partial",
    "affected": ["s3_storage"],
    "fallbacks": ["default_prompt", "skip_persistence"]
  }
}

// GET /health — Response
{
  "status": "degraded",
  "environment": "staging",
  "message": "Optional services unavailable: ['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}
  ]
}

---

Step-by-Step Implementation

Step 1: config/settings.py

"""config/settings.py — Settings for the 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 not allowed in production")
            if self.aws_endpoint_url:
                raise ValueError("aws_endpoint_url must not be set in production")
        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

Step 2: clients/factory.py

"""clients/factory.py — Factory for 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

Step 3: services/container.py

"""services/container.py — Service container with full DI."""

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,
        )

Step 4: handler.py

"""handler.py — Lambda handler for the 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 initialized: 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),
    }

Step 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

Step 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 — The operational artifact

# Migration Runbook: LocalStack → AWS Staging

## Prerequisites

- [ ] AWS account with IAM user/role configured
- [ ] AWS CLI configured (`aws sts get-caller-identity` works)
- [ ] Tests pass on LocalStack: `ENVIRONMENT=local pytest tests/ -v`
- [ ] .env.staging created with correct values

## Step 1: Verify staging config

```bash
# Validate that .env.staging has the correct values
python -c "
from config.settings import Settings
s = Settings(_env_file='.env.staging')
print(f'Environment: {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('✅ Staging config valid')
"

Step 2: Create AWS resources

# Create S3 bucket
aws s3 mb s3://ai-assets-staging-123456789012

# Verify
aws s3 ls s3://ai-assets-staging-123456789012

Step 3: Migrate prompt templates

# Export prompts from 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
"

# Import into AWS staging
ENVIRONMENT=staging python -c "
# ... import logic
"

Step 4: Run tests against staging

ENVIRONMENT=staging pytest tests/ -v --tb=short

# Expected:
# - Universal tests: PASSED
# - aws_only tests: PASSED
# - local_only tests: SKIPPED
# - sagemaker tests: PASSED (if endpoint deployed)

Step 5: Verify 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)')
"

Step 6: Functional smoke test

ENVIRONMENT=staging python -c "
from services.container import ServiceContainer
c = ServiceContainer.create()
result = c.processor.process(
    'summarizer', 'v1',
    {'id': 'smoke-test', 'content': 'Migration test'}
)
print(f'Result: {result}')
assert result['processed'] == True
print('✅ Smoke test passed')
"

Rollback

If something fails:

1. Change ENVIRONMENT=local → app returns to LocalStack
2. Do not delete AWS resources (so you can investigate)
3. Review logs: ENVIRONMENT=staging python -c "from clients.factory import ..."
4. Compare config: python config/compare.py local staging

Final Verification

• ENVIRONMENT=local pytest → all tests pass
• ENVIRONMENT=staging pytest → all tests pass
• Health endpoint returns "healthy" or "degraded" (not "unavailable")
• Functional smoke test passes
• Feature flags report correctly
• Rollback verified (switch to local and verify)

---

## Delivery Checklist

### Functionality

- [ ] The app runs with `ENVIRONMENT=local` against LocalStack
- [ ] The app runs with `ENVIRONMENT=staging` against AWS (or simulates successfully)
- [ ] `POST /process` processes documents and returns a result
- [ ] `GET /health` reports services, features, and degradation level
- [ ] Feature flags enable/disable SageMaker correctly
- [ ] Circuit breaker protects against S3 failures
- [ ] Fallback returns default prompts when S3 does not respond

### Architecture

- [ ] `config/settings.py` has Pydantic Settings with validation
- [ ] `clients/factory.py` has ClientFactory with client caching
- [ ] `services/container.py` has ServiceContainer with DI
- [ ] `services/feature_flags.py` has FeatureFlags with execute_if_enabled
- [ ] `services/circuit_breaker.py` has a working CircuitBreaker
- [ ] `handler.py` uses ServiceContainer (does not build clients directly)

### Testing

- [ ] `tests/conftest.py` detects the environment and configures fixtures
- [ ] Universal tests (S3, processor) pass in local
- [ ] aws_only tests are skipped in local, pass in staging
- [ ] Degradation tests verify fallbacks
- [ ] `pytest.ini` configured with markers

### Config

- [ ] `.env.local` configured for LocalStack
- [ ] `.env.staging` configured for AWS
- [ ] `.env.production` configured (even if not used yet)
- [ ] `.env.example` as a template
- [ ] `.gitignore` excludes .env files

### Documentation

- [ ] `migration/RUNBOOK.md` with concrete steps
- [ ] Runbook includes rollback
- [ ] Runbook includes verification at each step

---

## Project Verification

### Automated verification script

```python
"""verify_project.py — Verifies that the project meets all requirements."""

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 — VERIFICATION")
    print("=" * 60)

    results = []

    print("\n📁 Required files:")
    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📦 Importable modules:")
    modules = [
        "config.settings",
        "clients.factory",
        "services.container",
        "services.feature_flags",
    ]
    for m in modules:
        results.append(check_module_imports(m))

    print("\n🔧 Configuration:")
    try:
        from config.settings import Settings
        s = Settings()
        print(f"  ✅ Settings loads: env={s.environment}")
        results.append(True)
    except Exception as e:
        print(f"  ❌ Settings fails: {e}")
        results.append(False)

    print("\n📊 Result:")
    passed = sum(results)
    total = len(results)
    pct = passed / total * 100 if total > 0 else 0
    print(f"  {passed}/{total} checks passed ({pct:.0f}%)")

    if pct == 100:
        print("\n🎉 Project complete. Ready to migrate.")
    elif pct >= 80:
        print("\n⚠️ Almost ready. Review the failing items.")
    else:
        print("\n❌ Project incomplete. Review the checklist.")

    return pct == 100


if __name__ == "__main__":
    success = verify()
    sys.exit(0 if success else 1)

Running the verification

# Verify structure
python verify_project.py

# Tests against LocalStack
ENVIRONMENT=local pytest tests/ -v

# Tests against AWS (if available)
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}')
"

---

Connection with Later Modules

Module 7: Alternative Platforms

The abstraction layer you built here makes it easier to evaluate alternatives. If the M7 decision matrix says "Render instead of AWS," your migration-ready app can adapt:

M6 (here): ENVIRONMENT=local → LocalStack
            ENVIRONMENT=staging → AWS

M7: ENVIRONMENT=render → Render.com
    ENVIRONMENT=railway → Railway.app

The abstraction layer supports new environments
by adding config, not rewriting code.

Module 8: Integrative Project

The migration-ready app from M6 is the artifact that gets deployed end-to-end in M8:

M6: Build the migration-ready app (architecture)
    ↓
M7: Evaluate alternative platforms (decision)
    ↓
M8: Deploy to production with CI/CD + monitoring (operation)
    ├── The M6 app gets deployed
    ├── CI/CD uses the M6 tests
    ├── Monitoring uses the M6 health check
    └── Migration runbook runs in production

What you carry into M8

• ✅ App with environment abstraction (runs in any environment)
• ✅ Config management that supports multiple environments
• ✅ Test suite that verifies migration
• ✅ Health endpoint with degradation levels
• ✅ Feature flags for optional capabilities
• ✅ Documented migration runbook

---

Evaluation Criteria

Basic Level (pass)

• The app runs against LocalStack with ENVIRONMENT=local
• Config management with Pydantic Settings and .env files
• ClientFactory creates clients based on the environment
• At least 10 tests pass against LocalStack
• Migration runbook exists with concrete steps

Intermediate Level (well done)

• Everything from basic +
• Feature flags enable/disable features per environment
• Circuit breaker protects against S3 failures
• Health endpoint reports services and features
• Tests with markers (aws_only, local_only)
• 20+ tests pass

Advanced Level (excellent)

• Everything from intermediate +
• Full graceful degradation with cascading fallbacks
• Degradation tests that simulate service failures
• Config validation that rejects invalid configurations
• Dynamic feature flags (updatable at runtime)
• The app passes tests against LocalStack AND against AWS
• 30+ tests covering all the module's patterns

---

Summary

• This project is the most complete artifact of Phase 2. It integrates environment abstraction, config management, dependency injection, feature flags, graceful degradation, and multi-environment testing.
• Same code, any environment. Switching from LocalStack to AWS is changing one environment variable. The code, the tests, and the health check adapt automatically.
• The migration runbook is part of the deliverable. It's not just code — it's operational documentation another engineer can follow.
• This artifact carries into M8. The migration-ready app is what gets deployed to production with CI/CD, monitoring, and real operation in the Integrative Project.
• The patterns are transferable. Abstraction, DI, feature flags, circuit breakers — these patterns apply to any system, not just LocalStack/AWS. They're career skills.

---

Additional Resources

1. AWS Cloud Migration Best Practices — Official AWS migration guide
2. boto3 Documentation — Python SDK for AWS
3. LocalStack Documentation — Local development of AWS services
4. Pydantic Settings — Typed config management
5. AWS Well-Architected Framework — Architecture best practices
6. Migration Strategies for Python Applications — Lambda migration patterns

---