Module 4: LocalStack — AWS Local Development

6. Environment Switching

Next →

Overview

This capsule teaches you the most valuable skill of working with LocalStack: writing code that works against LocalStack in development and against real AWS in production — by changing only one environment variable. No environment if/else in your handler. No duplicated code. No conditional deployment logic. The same boto3, the same code, a different target.

Context: In the previous capsules you built an S3 + Lambda pipeline that works on LocalStack. Now you ask yourself: "how do I move this to real AWS without rewriting?" The answer is environment switching — and it's exactly what this capsule teaches. This concept is the basis of Module 6 (Cloud Migration Patterns), where you'll migrate from LocalStack to AWS with confidence.

The Core Concept

One variable changes everything

import boto3
import os

endpoint_url = os.environ.get("AWS_ENDPOINT_URL")

s3 = boto3.client(
    "s3",
    endpoint_url=endpoint_url,  # None → real AWS, URL → LocalStack
    region_name=os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
)

That's it. When AWS_ENDPOINT_URL doesn't exist (it's None), boto3 connects to real AWS. When it exists (for example, http://localhost:4566), it connects to LocalStack. Your code doesn't change. Not a single line.

Why it works

boto3 (the AWS SDK) has an endpoint_url parameter on all its clients. Normally, boto3 computes the endpoint automatically based on the service and the region (s3.us-east-1.amazonaws.com). But if you pass endpoint_url, boto3 uses that endpoint instead of the computed one.

LocalStack exposes the same APIs as AWS at localhost:4566. When boto3 sends a CreateBucket to localhost:4566, LocalStack processes the request just like AWS would. The request format, the headers, the payload — everything identical.

Development (LocalStack):
  boto3 → endpoint_url="http://localhost:4566" → LocalStack
  └── Same API, same methods, same format

Production (AWS):
  boto3 → endpoint_url=None → s3.us-east-1.amazonaws.com (AWS)
  └── Same API, same methods, same format

Code:
  IDENTICAL in both cases

Pattern 1: Simple Environment Variable

The most direct

# aws_config.py
import os
import boto3

def get_aws_client(service_name):
    """Creates an AWS client that points to LocalStack or AWS depending on the environment variable."""
    endpoint_url = os.environ.get("AWS_ENDPOINT_URL")

    kwargs = {
        "region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1"),
    }

    if endpoint_url:
        kwargs["endpoint_url"] = endpoint_url
        kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")

    return boto3.client(service_name, **kwargs)


# Usage — identical regardless of the environment
s3 = get_aws_client("s3")
lambda_client = get_aws_client("lambda")

Configuration per environment

# .env.development (LocalStack)
AWS_ENDPOINT_URL=http://localhost:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_DEFAULT_REGION=us-east-1

# .env.production (real AWS)
# AWS_ENDPOINT_URL is not defined → boto3 uses real AWS
AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=wJal...
AWS_DEFAULT_REGION=us-east-1
# .env.staging (LocalStack in CI/CD)
AWS_ENDPOINT_URL=http://localstack:4566
AWS_ACCESS_KEY_ID=test
AWS_SECRET_ACCESS_KEY=test
AWS_DEFAULT_REGION=us-east-1

The key: AWS_ENDPOINT_URL exists in dev and staging (LocalStack), doesn't exist in production (AWS).

Pattern 2: Config Class with Pydantic

For FastAPI applications

# config.py
import os
from pydantic_settings import BaseSettings
from typing import Optional


class AWSConfig(BaseSettings):
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: str = "test"
    aws_secret_access_key: str = "test"
    aws_default_region: str = "us-east-1"
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"
    model_name: str = "gpt-4o-mini"

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    @property
    def environment_name(self) -> str:
        if self.aws_endpoint_url:
            return "localstack"
        return "aws"

    class Config:
        env_file = ".env"


aws_config = AWSConfig()
# aws_clients.py
import boto3
from config import aws_config


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


def get_lambda_client():
    kwargs = {"region_name": aws_config.aws_default_region}
    if aws_config.aws_endpoint_url:
        kwargs["endpoint_url"] = aws_config.aws_endpoint_url
        kwargs["aws_access_key_id"] = aws_config.aws_access_key_id
        kwargs["aws_secret_access_key"] = aws_config.aws_secret_access_key
    return boto3.client("lambda", **kwargs)
# main.py — your app doesn't know or care whether it's LocalStack or AWS
from aws_clients import get_s3_client

s3 = get_s3_client()
s3.put_object(Bucket="ai-input", Key="test.txt", Body=b"hello")
# Works the same in LocalStack and AWS

Pattern 3: Factory with Logging

To know exactly where it points

# aws_factory.py
import boto3
import os
import logging

logger = logging.getLogger(__name__)


class AWSClientFactory:
    """Creates AWS clients with automatic environment switching."""

    def __init__(self):
        self.endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
        self.region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
        self._log_target()

    def _log_target(self):
        if self.endpoint_url:
            logger.info(f"AWS target: LocalStack ({self.endpoint_url})")
        else:
            logger.info(f"AWS target: real AWS (region={self.region})")

    def _base_kwargs(self):
        kwargs = {"region_name": self.region}
        if self.endpoint_url:
            kwargs["endpoint_url"] = self.endpoint_url
            kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
            kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
        return kwargs

    def s3(self):
        return boto3.client("s3", **self._base_kwargs())

    def lambda_client(self):
        return boto3.client("lambda", **self._base_kwargs())

    @property
    def is_local(self):
        return self.endpoint_url is not None


# Singleton
aws = AWSClientFactory()
# Usage
from aws_factory import aws

s3 = aws.s3()
s3.create_bucket(Bucket="my-bucket")

if aws.is_local:
    print("Operating against LocalStack — zero cost")
else:
    print("Operating against AWS — check costs")

Pattern for Lambda Handlers

The handler that works in both environments

# handler.py — works in LocalStack and AWS without changes
import json
import os
import boto3
from openai import OpenAI

AWS_ENDPOINT_URL = os.environ.get("AWS_ENDPOINT_URL")

openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", ""))

def _get_s3():
    kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
    if AWS_ENDPOINT_URL:
        kwargs["endpoint_url"] = AWS_ENDPOINT_URL
        kwargs["aws_access_key_id"] = "test"
        kwargs["aws_secret_access_key"] = "test"
    return boto3.client("s3", **kwargs)


def handler(event, context):
    s3 = _get_s3()

    # This code is IDENTICAL for LocalStack and AWS:
    doc = s3.get_object(Bucket="ai-input", Key=event["document_key"])
    text = doc["Body"].read().decode("utf-8")

    response = openai_client.chat.completions.create(
        model=os.environ.get("MODEL_NAME", "gpt-4o-mini"),
        messages=[
            {"role": "system", "content": "Analyze this document."},
            {"role": "user", "content": text},
        ],
        max_tokens=500,
    )

    result = {
        "analysis": response.choices[0].message.content,
        "tokens": response.usage.total_tokens,
        "environment": "localstack" if AWS_ENDPOINT_URL else "aws",
    }

    s3.put_object(
        Bucket="ai-output",
        Key=f"results/{event['document_key'].split('/')[-1]}.json",
        Body=json.dumps(result),
    )

    return {"statusCode": 200, "body": json.dumps(result)}

Environment variables per environment

# Deploy to LocalStack
awslocal lambda create-function \
  --function-name ai-processor \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    AWS_ENDPOINT_URL=http://host.docker.internal:4566,
    AWS_DEFAULT_REGION=us-east-1
  }" \
  ...

# Deploy to real AWS (future — Module 5)
aws lambda create-function \
  --function-name ai-processor \
  --environment "Variables={
    OPENAI_API_KEY=${OPENAI_API_KEY},
    MODEL_NAME=gpt-4o-mini,
    AWS_DEFAULT_REGION=us-east-1
  }" \
  ...
  # WITHOUT AWS_ENDPOINT_URL → boto3 uses real AWS automatically

Docker Compose for Different Environments

Base Compose + per-environment override

# docker-compose.yml (base — always used)
services:
  api:
    build: ./api
    ports:
      - "8000:8000"
    env_file:
      - .env
    depends_on:
      cache:
        condition: service_healthy

  cache:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 3
# docker-compose.localstack.yml (override for development with LocalStack)
services:
  api:
    environment:
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
    depends_on:
      localstack:
        condition: service_healthy

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
    volumes:
      - localstack_data:/var/lib/localstack
      - /var/run/docker.sock:/var/run/docker.sock
      - ./init-scripts:/etc/localstack/init/ready.d
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 15s

volumes:
  localstack_data:
# docker-compose.aws.yml (override for production with real AWS)
services:
  api:
    environment:
      - AWS_DEFAULT_REGION=us-east-1
      # AWS_ENDPOINT_URL is not defined → uses real AWS
      # Credentials come from the EC2/ECS IAM role
# Development with LocalStack
docker compose -f docker-compose.yml -f docker-compose.localstack.yml up -d

# Production with AWS
docker compose -f docker-compose.yml -f docker-compose.aws.yml up -d

Validate That the Switch Works

Verification script

# verify_switch.py
import boto3
import os

endpoint = os.environ.get("AWS_ENDPOINT_URL")
environment = "LocalStack" if endpoint else "AWS"

print(f"Environment: {environment}")
print(f"Endpoint: {endpoint or 'AWS default'}")
print(f"Region: {os.environ.get('AWS_DEFAULT_REGION', 'us-east-1')}")

# Create client
kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
if endpoint:
    kwargs["endpoint_url"] = endpoint
    kwargs["aws_access_key_id"] = "test"
    kwargs["aws_secret_access_key"] = "test"

s3 = boto3.client("s3", **kwargs)

# Test operation
try:
    buckets = s3.list_buckets()
    print(f"Buckets: {[b['Name'] for b in buckets['Buckets']]}")
    print(f"Connection successful to {environment}")
except Exception as e:
    print(f"Error connecting to {environment}: {e}")
# Test against LocalStack
AWS_ENDPOINT_URL=http://localhost:4566 python verify_switch.py
# Environment: LocalStack
# Buckets: ['ai-input', 'ai-output']
# Connection successful to LocalStack

# Test against AWS (if you have credentials)
unset AWS_ENDPOINT_URL
python verify_switch.py
# Environment: AWS
# Buckets: [...]  (your real buckets)
# Connection successful to AWS

Anti-patterns: What You Should NOT Do

Anti-pattern 1: environment if/else in the code

# BAD — don't do this
if os.environ.get("ENVIRONMENT") == "development":
    s3 = boto3.client("s3", endpoint_url="http://localhost:4566")
elif os.environ.get("ENVIRONMENT") == "staging":
    s3 = boto3.client("s3", endpoint_url="http://localstack:4566")
else:
    s3 = boto3.client("s3")

# GOOD — a single line
s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))

Anti-pattern 2: hardcoded URLs

# BAD
s3 = boto3.client("s3", endpoint_url="http://localhost:4566")

# GOOD
s3 = boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL"))

Anti-pattern 3: different code for each environment

# BAD — different handlers
# handler_local.py (for LocalStack)
# handler_aws.py (for AWS)

# GOOD — a single handler, the config controls it
# handler.py (works in both)

Anti-pattern 4: ignoring bucket naming

# BAD — hardcoded buckets
s3.put_object(Bucket="my-dev-bucket-123", Key="file.txt", Body=b"data")

# GOOD — buckets from config
BUCKET = os.environ.get("S3_INPUT_BUCKET", "ai-input")
s3.put_object(Bucket=BUCKET, Key="file.txt", Body=b"data")

Exercises

Exercise 1: Implement get_aws_client with logging

Create a get_aws_client function that accepts the service name, returns the correct boto3 client, and logs which environment it's connecting to (LocalStack or AWS).

See solution 
import boto3
import os
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("aws_client")


def get_aws_client(service_name):
    """Creates an AWS client with environment switching and logging."""
    endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
    region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")

    kwargs = {"region_name": region}

    if endpoint_url:
        kwargs["endpoint_url"] = endpoint_url
        kwargs["aws_access_key_id"] = os.environ.get("AWS_ACCESS_KEY_ID", "test")
        kwargs["aws_secret_access_key"] = os.environ.get("AWS_SECRET_ACCESS_KEY", "test")
        logger.info(f"[{service_name}] → LocalStack ({endpoint_url})")
    else:
        logger.info(f"[{service_name}] → AWS ({region})")

    return boto3.client(service_name, **kwargs)


# Test
s3 = get_aws_client("s3")
lambda_c = get_aws_client("lambda")

# With LocalStack:
# INFO:aws_client:[s3] → LocalStack (http://localhost:4566)
# INFO:aws_client:[lambda] → LocalStack (http://localhost:4566)

Exercise 2: Config class with validation

Create an AWSConfig class with Pydantic that validates the configuration: if aws_endpoint_url is defined, the credentials can be test. If it's not defined (real AWS), the credentials must start with AKIA (a real access key). Include an is_local property.

See solution 
from pydantic_settings import BaseSettings
from pydantic import model_validator
from typing import Optional


class AWSConfig(BaseSettings):
    aws_endpoint_url: Optional[str] = None
    aws_access_key_id: str = "test"
    aws_secret_access_key: str = "test"
    aws_default_region: str = "us-east-1"
    s3_input_bucket: str = "ai-input"
    s3_output_bucket: str = "ai-output"

    @property
    def is_local(self) -> bool:
        return self.aws_endpoint_url is not None

    @model_validator(mode="after")
    def validate_credentials(self):
        if not self.is_local:
            if not self.aws_access_key_id.startswith("AKIA"):
                raise ValueError(
                    "Real AWS requires valid credentials "
                    "(the access key must start with AKIA)"
                )
        return self

    class Config:
        env_file = ".env"


# Test with LocalStack (works with test/test)
import os
os.environ["AWS_ENDPOINT_URL"] = "http://localhost:4566"
config = AWSConfig()
print(f"Local: {config.is_local}")  # True

# Test without an endpoint (requires real credentials)
del os.environ["AWS_ENDPOINT_URL"]
os.environ["AWS_ACCESS_KEY_ID"] = "AKIA1234567890EXAMPLE"
os.environ["AWS_SECRET_ACCESS_KEY"] = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
config = AWSConfig()
print(f"Local: {config.is_local}")  # False

Exercise 3: Multi-environment Docker Compose

Create a base docker-compose.yml and a docker-compose.localstack.yml override. The base app has FastAPI. The override adds LocalStack and configures the environment variables. Demonstrate that the switch works by starting both.

See solution 
# docker-compose.yml (base)
services:
  api:
    build: ./api
    ports:
      - "8000:8000"
    environment:
      - AWS_DEFAULT_REGION=us-east-1
# docker-compose.localstack.yml (override)
services:
  api:
    environment:
      - AWS_ENDPOINT_URL=http://localstack:4566
      - AWS_ACCESS_KEY_ID=test
      - AWS_SECRET_ACCESS_KEY=test
    depends_on:
      localstack:
        condition: service_healthy

  localstack:
    image: localstack/localstack:latest
    ports:
      - "4566:4566"
    environment:
      - SERVICES=s3,lambda
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:4566/_localstack/health"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 15s
# Development with LocalStack
docker compose -f docker-compose.yml -f docker-compose.localstack.yml up -d

# Verify that api has the variables
docker compose exec api env | grep AWS
# AWS_ENDPOINT_URL=http://localstack:4566
# AWS_ACCESS_KEY_ID=test

# Production (base only — no LocalStack)
docker compose up -d
docker compose exec api env | grep AWS
# AWS_DEFAULT_REGION=us-east-1
# (no AWS_ENDPOINT_URL)

Exercise 4: LocalStack ↔ AWS compatibility test

Write a script that runs a sequence of S3 operations (create bucket, upload file, download, list, delete) and verifies that it works both against LocalStack and against AWS. The script should report success/failure for each operation.

See solution 
# test_compatibility.py
import boto3
import json
import os

endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
env_name = "LocalStack" if endpoint_url else "AWS"

kwargs = {"region_name": os.environ.get("AWS_DEFAULT_REGION", "us-east-1")}
if endpoint_url:
    kwargs["endpoint_url"] = endpoint_url
    kwargs["aws_access_key_id"] = "test"
    kwargs["aws_secret_access_key"] = "test"

s3 = boto3.client("s3", **kwargs)
bucket = "compatibility-test-bucket"
key = "test/data.json"
data = {"test": True, "environment": env_name}

results = {}

# 1. Create bucket
try:
    s3.create_bucket(Bucket=bucket)
    results["create_bucket"] = "PASS"
except Exception as e:
    results["create_bucket"] = f"FAIL: {e}"

# 2. Put object
try:
    s3.put_object(Bucket=bucket, Key=key, Body=json.dumps(data))
    results["put_object"] = "PASS"
except Exception as e:
    results["put_object"] = f"FAIL: {e}"

# 3. Get object
try:
    resp = s3.get_object(Bucket=bucket, Key=key)
    downloaded = json.loads(resp["Body"].read())
    assert downloaded == data
    results["get_object"] = "PASS"
except Exception as e:
    results["get_object"] = f"FAIL: {e}"

# 4. List objects
try:
    resp = s3.list_objects_v2(Bucket=bucket)
    assert resp["KeyCount"] == 1
    results["list_objects"] = "PASS"
except Exception as e:
    results["list_objects"] = f"FAIL: {e}"

# 5. Delete object
try:
    s3.delete_object(Bucket=bucket, Key=key)
    results["delete_object"] = "PASS"
except Exception as e:
    results["delete_object"] = f"FAIL: {e}"

# 6. Delete bucket
try:
    s3.delete_bucket(Bucket=bucket)
    results["delete_bucket"] = "PASS"
except Exception as e:
    results["delete_bucket"] = f"FAIL: {e}"

# Report
print(f"\n{'='*40}")
print(f"Compatibility Test — {env_name}")
print(f"{'='*40}")
passed = 0
for op, result in results.items():
    status = "✅" if result == "PASS" else "❌"
    print(f"  {status} {op}: {result}")
    if result == "PASS":
        passed += 1

print(f"\nResult: {passed}/{len(results)} operations successful")
# Test against LocalStack
AWS_ENDPOINT_URL=http://localhost:4566 python test_compatibility.py

# Test against AWS (if you have credentials)
unset AWS_ENDPOINT_URL
python test_compatibility.py

Troubleshooting

"My code works in LocalStack but fails in AWS"

# Most common cause: IAM permissions
# LocalStack Community doesn't enforce IAM
# AWS requires explicit permissions

# Solution: verify that your role/user has permissions for S3 and Lambda
aws iam list-attached-user-policies --user-name your-user

# For S3: you need s3:PutObject, s3:GetObject, s3:ListBucket
# For Lambda: you need lambda:InvokeFunction

"The 'test' credentials don't work in AWS"

# Correct — 'test/test' is for LocalStack
# AWS needs real credentials
aws configure
# Enter your real credentials (AKIA...)

"endpoint_url=None causes an error"

# boto3 accepts endpoint_url=None — it ignores it and uses the default
# If you get an error, verify you're not passing an empty string:

# BAD
endpoint = os.environ.get("AWS_ENDPOINT_URL", "")
# If AWS_ENDPOINT_URL doesn't exist, endpoint = "" (empty string, not None)

# GOOD
endpoint = os.environ.get("AWS_ENDPOINT_URL")
# If it doesn't exist, endpoint = None (boto3 handles it correctly)

Summary

  • Environment switching = the same code points to LocalStack or AWS by changing only AWS_ENDPOINT_URL.
  • When AWS_ENDPOINT_URL is None, boto3 connects to real AWS. When it has a value, it connects to that endpoint (LocalStack).
  • The cleanest pattern: boto3.client("s3", endpoint_url=os.environ.get("AWS_ENDPOINT_URL")). One line.
  • Docker Compose overrides let you start up with or without LocalStack depending on the environment.
  • Anti-patterns to avoid: environment if/else, hardcoded URLs, code duplicated per environment.
  • This pattern is the basis of migration in Module 6: if your code works with environment switching, migrating to AWS is changing a variable.
  • Remember: LocalStack Community doesn't enforce IAM. In real AWS, verify permissions.

Additional Resources

  1. boto3 Configuration — How boto3 resolves credentials and endpoints
  2. AWS Environment Variables — Official AWS environment variables
  3. 12-Factor App — Config — The philosophy of config via environment variables
  4. Pydantic Settings — Settings management with Pydantic
  5. Docker Compose Override — How overrides work in Compose
  6. LocalStack AWS Feature Coverage — What works the same and what differs
  7. AWS IAM Best Practices — For when you migrate to real AWS
← Previous5. S3 + Lambda Local PipelineNext →7. Debugging LocalStack