Module 6: Cloud Migration Patterns

2. Environment Abstraction

Next →

Overview

In this capsule you'll learn the most fundamental pattern of cloud migration: environment abstraction. It's the ability for your code to interact with LocalStack or AWS without knowing which is which. The code doesn't ask "am I in development or production?" — it receives configuration and runs. The difference between a junior developer who hardcodes endpoint_url="http://localhost:4566" and a senior engineer who abstracts the environment is exactly this pattern.

Context: In Modules 4 and 5, your code has endpoint_url hardcoded in some files and absent in others. Switching from LocalStack to AWS requires editing lines of code. That works for a personal project, but in an engineering team with 3+ environments (dev, staging, prod), it doesn't scale. A mistake in one endpoint_url connects you to the wrong environment — and you write test data into production. Environment abstraction eliminates that class of errors by design.

The Problem: Code Coupled to the Environment

Real case: the M4 and M5 code

Look at how boto3 clients are created in the previous modules:

# Module 4 — code coupled to LocalStack
import boto3

s3 = boto3.client(
    "s3",
    endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    region_name="us-east-1",
)

bucket = "ai-assets-dev"
s3.put_object(Bucket=bucket, Key="test.txt", Body=b"hello")
# Module 5 — code coupled to AWS
import boto3

s3 = boto3.client("s3")  # Uses ~/.aws/credentials

bucket = "ai-assets-123456789012"
s3.put_object(Bucket=bucket, Key="test.txt", Body=b"hello")

Visible problem: these are two different files to do the same thing against two environments. If you want to run the M4 tests against AWS, you need to rewrite the client creation. If you want to test the M5 code on LocalStack, you need to add endpoint_url.

The 4 elements that change between environments

┌─────────────────────────────────────────────────────┐
│          What changes between environments           │
├──────────────────────┬───────────────────────────────┤
│ 1. Endpoint URL      │ localhost:4566 vs AWS default │
│ 2. Credentials       │ test/test vs IAM roles        │
│ 3. Resource names    │ ai-dev vs ai-prod-123456      │
│ 4. Features          │ SageMaker not on LocalStack   │
└──────────────────────┴───────────────────────────────┘

┌─────────────────────────────────────────────────────┐
│       What does NOT change between environments      │
├─────────────────────────────────────────────────────┤
│ - Business logic (process document, invoke LLM)     │
│ - S3 key structure (prompts/, documents/, etc.)     │
│ - Request/response format                           │
│ - Error handling                                    │
│ - Functional tests                                  │
└─────────────────────────────────────────────────────┘

The rule is simple: what changes between environments goes in configuration. What doesn't change goes in code.

The Pattern: Environment Abstraction

Level 1 — Simple environment variable

The first level of abstraction is an ENVIRONMENT variable that decides how to create the clients:

import boto3
import os


def get_s3_client():
    """Creates an S3 client configured for the current environment."""
    environment = os.environ.get("ENVIRONMENT", "local")

    if environment == "local":
        return boto3.client(
            "s3",
            endpoint_url="http://localhost:4566",
            aws_access_key_id="test",
            aws_secret_access_key="test",
            region_name="us-east-1",
        )

    # AWS (staging or production)
    return boto3.client("s3", region_name=os.environ.get("AWS_REGION", "us-east-1"))

This works, but it has problems:

  • ❌ The if/else grows with every new environment
  • ❌ The values are hardcoded inside the function
  • ❌ There's no validation that ENVIRONMENT is a valid value
  • ❌ It's not testable (you can't inject configuration)

Level 2 — Config dict with endpoint switching

A step better: the configuration lives in a dictionary and the client is created from it:

import boto3
import os
from typing import Optional


ENVIRONMENT_CONFIGS = {
    "local": {
        "endpoint_url": "http://localhost:4566",
        "aws_access_key_id": "test",
        "aws_secret_access_key": "test",
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-local",
    },
    "staging": {
        "endpoint_url": None,
        "aws_access_key_id": None,  # Uses IAM role
        "aws_secret_access_key": None,
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-staging-123456",
    },
    "production": {
        "endpoint_url": None,
        "aws_access_key_id": None,
        "aws_secret_access_key": None,
        "region_name": "us-east-1",
        "s3_bucket": "ai-assets-prod-123456",
    },
}


def get_environment_config(env_name: Optional[str] = None) -> dict:
    """Returns the configuration for the specified environment."""
    env = env_name or os.environ.get("ENVIRONMENT", "local")
    if env not in ENVIRONMENT_CONFIGS:
        raise ValueError(
            f"Environment '{env}' not recognized. "
            f"Options: {list(ENVIRONMENT_CONFIGS.keys())}"
        )
    return ENVIRONMENT_CONFIGS[env]


def create_s3_client(config: dict):
    """Creates an S3 client from a config dict."""
    client_kwargs = {"region_name": config["region_name"]}
    if config.get("endpoint_url"):
        client_kwargs["endpoint_url"] = config["endpoint_url"]
    if config.get("aws_access_key_id"):
        client_kwargs["aws_access_key_id"] = config["aws_access_key_id"]
        client_kwargs["aws_secret_access_key"] = config["aws_secret_access_key"]
    return boto3.client("s3", **client_kwargs)

Improvements:

  • ✅ The values aren't in the client-creation code
  • ✅ Adding a new environment is adding an entry to the dict
  • ✅ The client is created from config, not from logic

But it still has problems:

  • ❌ The config is in the source code (not in .env files)
  • ❌ There's no typing or validation
  • ❌ The secrets are visible in the dict

Level 3 — The complete pattern (what you'll build)

The mature pattern uses Pydantic Settings to type and validate, loads from .env files, and a factory creates clients:

from pydantic_settings import BaseSettings
from typing import Optional
import boto3


class EnvironmentSettings(BaseSettings):
    """Typed environment configuration."""
    environment: str = "local"
    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"

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"


def create_boto3_client(service: str, settings: EnvironmentSettings):
    """Creates a boto3 client configured according to the environment."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key
    return boto3.client(service, **kwargs)

With per-environment .env files:

# .env.local
ENVIRONMENT=local
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
# .env.staging
ENVIRONMENT=staging
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-staging-123456
# Credentials via IAM role, not in .env
# .env.production
ENVIRONMENT=production
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-prod-123456
# Credentials via IAM role, not in .env

Now the business code is completely agnostic:

settings = EnvironmentSettings()  # Reads from .env
s3 = create_boto3_client("s3", settings)

def process_document(doc_key: str) -> dict:
    """Processes a document — works in any environment."""
    response = s3.get_object(Bucket=settings.s3_bucket, Key=doc_key)
    content = response["Body"].read().decode("utf-8")
    # ... identical business logic ...
    return {"status": "processed", "key": doc_key}

Step-by-Step Implementation

Step 1: Create the configuration class

"""config/environment.py — Environment configuration."""

from pydantic_settings import BaseSettings
from pydantic import field_validator
from typing import Optional
from enum import Enum


class EnvironmentName(str, Enum):
    LOCAL = "local"
    STAGING = "staging"
    PRODUCTION = "production"


class EnvironmentSettings(BaseSettings):
    """Environment configuration with validation."""

    environment: EnvironmentName = EnvironmentName.LOCAL
    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"

    @field_validator("environment", mode="before")
    @classmethod
    def validate_environment(cls, v):
        if isinstance(v, str):
            v = v.lower().strip()
        return v

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

    @property
    def is_production(self) -> bool:
        return self.environment == EnvironmentName.PRODUCTION

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        use_enum_values = True

Step 2: Function to create generic clients

"""clients/base.py — Creating environment-agnostic boto3 clients."""

import boto3
from config.environment import EnvironmentSettings


def create_client(service: str, settings: EnvironmentSettings):
    """Creates a boto3 client configured for the current environment.

    In local (LocalStack): uses endpoint_url and test credentials.
    In AWS (staging/prod): uses the IAM role (no endpoint_url).
    """
    kwargs = {"region_name": settings.aws_region}

    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url

    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    return boto3.client(service, **kwargs)


def create_s3_client(settings: EnvironmentSettings):
    return create_client("s3", settings)


def create_lambda_client(settings: EnvironmentSettings):
    return create_client("lambda", settings)

Step 3: Verify it works in both environments

"""verify_abstraction.py — Verifies that the same code works in both environments."""

import os
from config.environment import EnvironmentSettings
from clients.base import create_s3_client


def verify_environment():
    """Verifies connectivity to the configured environment."""
    settings = EnvironmentSettings()
    print(f"Environment: {settings.environment}")
    print(f"Bucket: {settings.s3_bucket}")
    print(f"Endpoint: {settings.aws_endpoint_url or 'AWS default'}")
    print(f"Region: {settings.aws_region}")
    print()

    s3 = create_s3_client(settings)

    try:
        s3.head_bucket(Bucket=settings.s3_bucket)
        print(f"✅ Bucket '{settings.s3_bucket}' accessible")
    except s3.exceptions.ClientError as e:
        error_code = e.response["Error"]["Code"]
        if error_code == "404":
            print(f"⚠️ Bucket '{settings.s3_bucket}' doesn't exist. Creating it...")
            s3.create_bucket(Bucket=settings.s3_bucket)
            print(f"✅ Bucket created")
        else:
            print(f"❌ Error: {e}")
            return False
    except Exception as e:
        print(f"❌ Cannot connect to the environment: {e}")
        return False

    test_key = "_health/connection-test.txt"
    s3.put_object(
        Bucket=settings.s3_bucket,
        Key=test_key,
        Body=b"connection test OK",
    )
    response = s3.get_object(Bucket=settings.s3_bucket, Key=test_key)
    content = response["Body"].read().decode("utf-8")
    assert content == "connection test OK"
    s3.delete_object(Bucket=settings.s3_bucket, Key=test_key)
    print(f"✅ Read/write verified on '{settings.s3_bucket}'")

    return True


if __name__ == "__main__":
    verify_environment()

Execution:

# Against LocalStack
ENVIRONMENT=local python verify_abstraction.py
# Output:
# Environment: local
# Bucket: ai-assets-local
# Endpoint: http://localhost:4566
# ✅ Bucket 'ai-assets-local' accessible
# ✅ Read/write verified

# Against AWS (if you have an account)
ENVIRONMENT=staging python verify_abstraction.py
# Output:
# Environment: staging
# Bucket: ai-assets-staging-123456
# Endpoint: AWS default
# ✅ Bucket 'ai-assets-staging-123456' accessible
# ✅ Read/write verified

Same script, same code, different environments. The ENVIRONMENT variable and the corresponding .env file make all the difference.

The Transparent Pattern: boto3 Doesn't Need to Know

What boto3 really needs

boto3 operates on a simple principle: if you give it endpoint_url, it talks to that endpoint. If not, it talks to real AWS. Everything else (credentials, region) works the same:

import boto3

# These two clients are functionally equivalent
# for S3 operations:

# LocalStack client
local = boto3.client("s3", endpoint_url="http://localhost:4566",
                     aws_access_key_id="test",
                     aws_secret_access_key="test")

# AWS client
aws = boto3.client("s3")

# The same operation works with both:
# local.put_object(Bucket="b", Key="k", Body=b"data")
# aws.put_object(Bucket="b", Key="k", Body=b"data")

Endpoint switching is the mechanism that makes the whole abstraction possible. You don't need complex wrappers or enterprise abstractions — just control which parameters boto3.client() receives.

Complete pattern: environment-agnostic business logic

"""services/document_processor.py — 100% environment-agnostic business logic."""

import json
from datetime import datetime


class DocumentProcessor:
    """Processes AI documents. It doesn't know or care about the environment."""

    def __init__(self, s3_client, bucket_name: str):
        self.s3 = s3_client
        self.bucket = bucket_name

    def get_prompt_template(self, name: str, version: str) -> str:
        key = f"prompts/{name}/{version}/system.txt"
        response = self.s3.get_object(Bucket=self.bucket, Key=key)
        return response["Body"].read().decode("utf-8")

    def store_document(self, collection: str, doc_id: str, data: dict) -> str:
        key = f"documents/{collection}/{doc_id}.json"
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(data, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def save_response(self, request_id: str, result: dict) -> str:
        now = datetime.utcnow()
        key = f"responses/{now.strftime('%Y/%m/%d')}/{request_id}.json"
        payload = {
            "request_id": request_id,
            "timestamp": now.isoformat(),
            "result": result,
        }
        self.s3.put_object(
            Bucket=self.bucket,
            Key=key,
            Body=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
            ContentType="application/json",
        )
        return key

    def process(self, prompt_name: str, prompt_version: str, document: dict) -> dict:
        template = self.get_prompt_template(prompt_name, prompt_version)
        stored_key = self.store_document(
            "inbox", document.get("id", "unknown"), document
        )
        result = {
            "prompt_used": f"{prompt_name}/{prompt_version}",
            "document_key": stored_key,
            "processed": True,
            "summary": f"Processed with template: {template[:50]}...",
        }
        response_key = self.save_response(
            f"req-{datetime.utcnow().strftime('%H%M%S')}", result
        )
        result["response_key"] = response_key
        return result

Usage in any environment:

from config.environment import EnvironmentSettings
from clients.base import create_s3_client
from services.document_processor import DocumentProcessor

settings = EnvironmentSettings()
s3 = create_s3_client(settings)
processor = DocumentProcessor(s3, settings.s3_bucket)

result = processor.process(
    prompt_name="summarizer",
    prompt_version="v1",
    document={"id": "doc-001", "content": "Document text..."},
)
print(result)

DocumentProcessor doesn't import os, doesn't read environment variables, doesn't have if local/aws. It receives an S3 client and a bucket name — and it works.

Troubleshooting

Problem 1: "Could not connect to the endpoint URL" with LocalStack

LocalStack isn't running or the port is different.

# Verify that LocalStack is running
docker ps | grep localstack

# If it doesn't appear:
docker run -d --name localstack -p 4566:4566 localstack/localstack

# Verify connectivity
curl http://localhost:4566/_localstack/health

Problem 2: The client connects to AWS when it should go to LocalStack

The ENVIRONMENT variable isn't set or the .env isn't loaded.

import os
print(f"ENVIRONMENT = {os.environ.get('ENVIRONMENT', 'NOT DEFINED')}")

# If it's empty, Pydantic Settings uses the default ("local")
# But if you have AWS credentials in ~/.aws, boto3 might use AWS anyway

# Solution: make sure .env.local has AWS_ENDPOINT_URL

Problem 3: "Invalid endpoint" when switching from local to staging

The staging .env has AWS_ENDPOINT_URL set (it shouldn't).

# .env.staging — CORRECT
ENVIRONMENT=staging
AWS_REGION=us-east-1
S3_BUCKET=ai-assets-staging-123456
# No AWS_ENDPOINT_URL → boto3 uses AWS default

# .env.staging — INCORRECT
ENVIRONMENT=staging
AWS_ENDPOINT_URL=http://localhost:4566  # ← This connects to LocalStack, not AWS

Problem 4: Pydantic doesn't load the correct .env file

Pydantic Settings loads .env by default. To load .env.local or .env.staging, you need to specify it:

from pydantic_settings import BaseSettings
import os

env_name = os.environ.get("ENVIRONMENT", "local")

class Settings(BaseSettings):
    class Config:
        env_file = f".env.{env_name}"

Or use the ENVIRONMENT variable directly as a system env var, and the .env only for additional values.

Practical Exercises

Exercise 1: Multi-service client factory

Create a create_clients function that takes an EnvironmentSettings and returns a dictionary with clients for S3, Lambda, and CloudWatch Logs — all configured for the same environment.

See solution 
import boto3
from config.environment import EnvironmentSettings


def create_clients(settings: EnvironmentSettings) -> dict:
    """Creates clients for multiple AWS services in the current environment."""
    services = ["s3", "lambda", "logs"]
    clients = {}

    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    for service in services:
        clients[service] = boto3.client(service, **kwargs)

    return clients


settings = EnvironmentSettings()
clients = create_clients(settings)

print(f"Environment: {settings.environment}")
print(f"Clients created: {list(clients.keys())}")

buckets = clients["s3"].list_buckets()
print(f"Accessible buckets: {len(buckets.get('Buckets', []))}")

Exercise 2: Environment health check

Implement a check_environment_health function that verifies connectivity to S3 and Lambda for the current environment, returning a dict with the status of each service.

See solution 
import boto3
from config.environment import EnvironmentSettings


def check_environment_health(settings: EnvironmentSettings) -> dict:
    """Verifies the health of the services in the current environment."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    health = {
        "environment": settings.environment,
        "services": {},
        "overall": "healthy",
    }

    # Check S3
    try:
        s3 = boto3.client("s3", **kwargs)
        s3.list_buckets()
        health["services"]["s3"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["s3"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    # Check Lambda
    try:
        lam = boto3.client("lambda", **kwargs)
        lam.list_functions(MaxItems=1)
        health["services"]["lambda"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["lambda"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    # Check bucket access
    try:
        s3 = boto3.client("s3", **kwargs)
        s3.head_bucket(Bucket=settings.s3_bucket)
        health["services"]["s3_bucket"] = {"status": "healthy", "error": None}
    except Exception as e:
        health["services"]["s3_bucket"] = {"status": "unhealthy", "error": str(e)}
        health["overall"] = "degraded"

    return health


settings = EnvironmentSettings()
result = check_environment_health(settings)
print(f"Environment: {result['environment']}")
print(f"Overall status: {result['overall']}")
for service, info in result["services"].items():
    status_icon = "✅" if info["status"] == "healthy" else "❌"
    print(f"  {status_icon} {service}: {info['status']}")

Exercise 3: Environment comparison report

Create a script that connects to two environments (local and staging) and compares which buckets and Lambda functions exist in each, generating a report of the differences.

See solution 
import boto3
from config.environment import EnvironmentSettings, EnvironmentName


def get_environment_resources(settings: EnvironmentSettings) -> dict:
    """Lists the resources available in an environment."""
    kwargs = {"region_name": settings.aws_region}
    if settings.aws_endpoint_url:
        kwargs["endpoint_url"] = settings.aws_endpoint_url
    if settings.aws_access_key_id:
        kwargs["aws_access_key_id"] = settings.aws_access_key_id
        kwargs["aws_secret_access_key"] = settings.aws_secret_access_key

    resources = {"buckets": [], "functions": []}

    try:
        s3 = boto3.client("s3", **kwargs)
        response = s3.list_buckets()
        resources["buckets"] = [b["Name"] for b in response.get("Buckets", [])]
    except Exception as e:
        resources["buckets_error"] = str(e)

    try:
        lam = boto3.client("lambda", **kwargs)
        response = lam.list_functions()
        resources["functions"] = [
            f["FunctionName"] for f in response.get("Functions", [])
        ]
    except Exception as e:
        resources["functions_error"] = str(e)

    return resources


def compare_environments():
    """Compares resources between the local and staging environments."""
    local_settings = EnvironmentSettings(
        environment=EnvironmentName.LOCAL,
        aws_endpoint_url="http://localhost:4566",
        aws_access_key_id="test",
        aws_secret_access_key="test",
        s3_bucket="ai-assets-local",
    )

    local_res = get_environment_resources(local_settings)

    print("=" * 60)
    print("ENVIRONMENT COMPARISON REPORT")
    print("=" * 60)

    print(f"\n📍 LOCAL (LocalStack)")
    print(f"  Buckets: {local_res['buckets'] or 'none'}")
    print(f"  Functions: {local_res['functions'] or 'none'}")

    # For staging, you'd need real AWS credentials
    # Here we show the pattern:
    print(f"\n📍 STAGING (AWS)")
    print(f"  (Requires configured AWS credentials)")

    print(f"\n📊 DIFFERENCES:")
    print(f"  Resources only in local: {len(local_res['buckets'])} buckets")
    print(f"  The full report requires access to both environments")


compare_environments()

Exercise 4: Migrating assets between environments

Implement a function that copies all the prompt templates from one environment (LocalStack) to another (AWS staging), verifying that each template was copied correctly.

See solution 
import boto3
from config.environment import EnvironmentSettings, EnvironmentName


def migrate_prompts(
    source_settings: EnvironmentSettings,
    target_settings: EnvironmentSettings,
    dry_run: bool = True,
) -> dict:
    """Migrates prompt templates from one environment to another."""

    def make_client(settings):
        kwargs = {"region_name": settings.aws_region}
        if settings.aws_endpoint_url:
            kwargs["endpoint_url"] = settings.aws_endpoint_url
        if settings.aws_access_key_id:
            kwargs["aws_access_key_id"] = settings.aws_access_key_id
            kwargs["aws_secret_access_key"] = settings.aws_secret_access_key
        return boto3.client("s3", **kwargs)

    source_s3 = make_client(source_settings)
    target_s3 = make_client(target_settings)

    paginator = source_s3.get_paginator("list_objects_v2")
    results = {"migrated": [], "failed": [], "skipped": []}

    for page in paginator.paginate(
        Bucket=source_settings.s3_bucket, Prefix="prompts/"
    ):
        for obj in page.get("Contents", []):
            key = obj["Key"]
            if dry_run:
                results["skipped"].append(key)
                print(f"  [DRY RUN] Would migrate: {key}")
                continue

            try:
                response = source_s3.get_object(
                    Bucket=source_settings.s3_bucket, Key=key
                )
                body = response["Body"].read()

                target_s3.put_object(
                    Bucket=target_settings.s3_bucket,
                    Key=key,
                    Body=body,
                    ContentType=response.get("ContentType", "application/octet-stream"),
                )

                # Verify
                check = target_s3.get_object(
                    Bucket=target_settings.s3_bucket, Key=key
                )
                check_body = check["Body"].read()

                if body == check_body:
                    results["migrated"].append(key)
                    print(f"  ✅ Migrated and verified: {key}")
                else:
                    results["failed"].append(key)
                    print(f"  ❌ Content doesn't match: {key}")

            except Exception as e:
                results["failed"].append(key)
                print(f"  ❌ Error migrating {key}: {e}")

    print(f"\nResult: {len(results['migrated'])} migrated, "
          f"{len(results['failed'])} failed, "
          f"{len(results['skipped'])} skipped (dry_run)")
    return results


source = EnvironmentSettings(
    environment=EnvironmentName.LOCAL,
    aws_endpoint_url="http://localhost:4566",
    aws_access_key_id="test",
    aws_secret_access_key="test",
    s3_bucket="ai-assets-local",
)

# For a real migration, target would be AWS staging
# target = EnvironmentSettings(environment="staging", s3_bucket="ai-assets-staging-xxx")

# Dry run against the same environment (for demo)
migrate_prompts(source, source, dry_run=True)

Summary

  • Environment abstraction is the most fundamental cloud migration pattern. Without it, every environment change requires editing source code — unsustainable in teams and dangerous in production.
  • What changes between environments goes in configuration, what doesn't goes in code. Endpoints, credentials, bucket names → config. Business logic → code.
  • Three maturity levels: Simple variable → config dict → Pydantic Settings with .env files. This module takes you to level 3.
  • boto3 is naturally abstract. It only needs endpoint_url for LocalStack; without it, it talks to AWS. Endpoint switching is all you need.
  • DocumentProcessor is the example to follow. It receives a client and a bucket name. It doesn't know the environment. It doesn't care.
  • In the next capsule, you'll go deeper into config management — Pydantic Settings, validation, secrets, and per-environment .env files.

Additional Resources

  1. The Twelve-Factor App — Config — The principle: config in the environment, not in the code
  2. boto3 Configuration — How boto3 resolves configuration
  3. Pydantic Settings — Typed settings management
  4. LocalStack Configuration — Configuring LocalStack
  5. AWS SDK Configuration — AWS SDK configuration reference
  6. Environment Variables Best Practices — Principles of external configuration
← Previous1. Introduction: Cloud Migration PatternsNext →3. Multi-environment Config Management