Configuration Management: Environment Variables, Config Maps, and 12-Factor Apps

import os
from dataclasses import dataclass, field
from typing import Optional, List
import logging

@dataclass
class DatabaseConfig:
    """Database configuration."""
    url: str
    pool_size: int = 10
    pool_timeout: int = 30
    echo_sql: bool = False

@dataclass
class RedisConfig:
    """Redis configuration."""
    url: str
    max_connections: int = 50
    socket_timeout: int = 5

@dataclass
class AppConfig:
    """Application configuration from environment variables."""
    
    # Required configuration
    database: DatabaseConfig
    redis: RedisConfig
    secret_key: str
    
    # Optional with defaults
    environment: str = "production"
    log_level: str = "INFO"
    debug: bool = False
    port: int = 8000
    workers: int = 4
    allowed_hosts: List[str] = field(default_factory=lambda: ["*"])
    cors_origins: List[str] = field(default_factory=list)
    
    # Feature flags
    enable_caching: bool = True
    enable_metrics: bool = True
    enable_tracing: bool = False
    
    @classmethod
    def from_env(cls) -> "AppConfig":
        """Load configuration from environment variables."""
        
        # Required variables
        database_url = os.getenv("DATABASE_URL")
        if not database_url:
            raise ValueError("DATABASE_URL is required")
        
        redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
        secret_key = os.getenv("SECRET_KEY")
        if not secret_key:
            raise ValueError("SECRET_KEY is required")
        
        # Database config
        database = DatabaseConfig(
            url=database_url,
            pool_size=int(os.getenv("DB_POOL_SIZE", "10")),
            pool_timeout=int(os.getenv("DB_POOL_TIMEOUT", "30")),
            echo_sql=os.getenv("DB_ECHO_SQL", "false").lower() == "true"
        )
        
        # Redis config
        redis = RedisConfig(
            url=redis_url,
            max_connections=int(os.getenv("REDIS_MAX_CONNECTIONS", "50")),
            socket_timeout=int(os.getenv("REDIS_TIMEOUT", "5"))
        )
        
        # Parse list values
        allowed_hosts = os.getenv("ALLOWED_HOSTS", "*").split(",")
        cors_origins = os.getenv("CORS_ORIGINS", "").split(",") if os.getenv("CORS_ORIGINS") else []
        
        return cls(
            database=database,
            redis=redis,
            secret_key=secret_key,
            environment=os.getenv("ENVIRONMENT", "production"),
            log_level=os.getenv("LOG_LEVEL", "INFO"),
            debug=os.getenv("DEBUG", "false").lower() == "true",
            port=int(os.getenv("PORT", "8000")),
            workers=int(os.getenv("WORKERS", "4")),
            allowed_hosts=allowed_hosts,
            cors_origins=cors_origins,
            enable_caching=os.getenv("ENABLE_CACHING", "true").lower() == "true",
            enable_metrics=os.getenv("ENABLE_METRICS", "true").lower() == "true",
            enable_tracing=os.getenv("ENABLE_TRACING", "false").lower() == "true"
        )
    
    def setup_logging(self):
        """Configure logging based on config."""
        logging.basicConfig(
            level=getattr(logging, self.log_level),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )

# Usage
config = AppConfig.from_env()
config.setup_logging()

package config

import (
    "fmt"
    "os"
    "strconv"
    "strings"
    "time"
)

type Config struct {
    // Server
    Port         int
    Environment  string
    Debug        bool
    
    // Database
    DatabaseURL  string
    DBPoolSize   int
    DBTimeout    time.Duration
    
    // Redis
    RedisURL     string
    RedisTimeout time.Duration
    
    // Security
    SecretKey    string
    JWTSecret    string
    
    // Features
    EnableCache  bool
    EnableMetrics bool
}

func LoadFromEnv() (*Config, error) {
    cfg := &Config{
        Port:         getEnvAsInt("PORT", 8080),
        Environment:  getEnv("ENVIRONMENT", "production"),
        Debug:        getEnvAsBool("DEBUG", false),
        
        DatabaseURL:  getEnv("DATABASE_URL", ""),
        DBPoolSize:   getEnvAsInt("DB_POOL_SIZE", 10),
        DBTimeout:    getEnvAsDuration("DB_TIMEOUT", 30*time.Second),
        
        RedisURL:     getEnv("REDIS_URL", "redis://localhost:6379"),
        RedisTimeout: getEnvAsDuration("REDIS_TIMEOUT", 5*time.Second),
        
        SecretKey:    getEnv("SECRET_KEY", ""),
        JWTSecret:    getEnv("JWT_SECRET", ""),
        
        EnableCache:  getEnvAsBool("ENABLE_CACHE", true),
        EnableMetrics: getEnvAsBool("ENABLE_METRICS", true),
    }
    
    // Validate required fields
    if cfg.DatabaseURL == "" {
        return nil, fmt.Errorf("DATABASE_URL is required")
    }
    if cfg.SecretKey == "" {
        return nil, fmt.Errorf("SECRET_KEY is required")
    }
    
    return cfg, nil
}

func getEnv(key, defaultValue string) string {
    if value := os.Getenv(key); value != "" {
        return value
    }
    return defaultValue
}

func getEnvAsInt(key string, defaultValue int) int {
    if value := os.Getenv(key); value != "" {
        if intVal, err := strconv.Atoi(value); err == nil {
            return intVal
        }
    }
    return defaultValue
}

func getEnvAsBool(key string, defaultValue bool) bool {
    if value := os.Getenv(key); value != "" {
        return strings.ToLower(value) == "true"
    }
    return defaultValue
}

func getEnvAsDuration(key string, defaultValue time.Duration) time.Duration {
    if value := os.Getenv(key); value != "" {
        if duration, err := time.ParseDuration(value); err == nil {
            return duration
        }
    }
    return defaultValue
}

// config.js
const dotenv = require('dotenv');
dotenv.config();

class Config {
  constructor() {
    // Required
    this.databaseUrl = this.required('DATABASE_URL');
    this.secretKey = this.required('SECRET_KEY');
    
    // Optional with defaults
    this.environment = process.env.ENVIRONMENT || 'production';
    this.port = parseInt(process.env.PORT || '3000', 10);
    this.logLevel = process.env.LOG_LEVEL || 'info';
    this.debug = process.env.DEBUG === 'true';
    
    // Redis
    this.redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';
    this.redisTimeout = parseInt(process.env.REDIS_TIMEOUT || '5000', 10);
    
    // Features
    this.enableCache = process.env.ENABLE_CACHE !== 'false';
    this.enableMetrics = process.env.ENABLE_METRICS !== 'false';
    
    // Lists
    this.allowedHosts = (process.env.ALLOWED_HOSTS || '*').split(',');
    this.corsOrigins = process.env.CORS_ORIGINS 
      ? process.env.CORS_ORIGINS.split(',') 
      : [];
  }
  
  required(key) {
    const value = process.env[key];
    if (!value) {
      throw new Error(`${key} is required`);
    }
    return value;
  }
  
  isDevelopment() {
    return this.environment === 'development';
  }
  
  isProduction() {
    return this.environment === 'production';
  }
}

module.exports = new Config();

# Single repository
my-app/
├── .git/
├── src/
├── tests/
├── Dockerfile
└── README.md

# Multiple deployments from same codebase
git push heroku main          # Deploy to production
git push staging main         # Deploy to staging
git push development main     # Deploy to development

# requirements.txt
fastapi==0.109.0
uvicorn[standard]==0.27.0
sqlalchemy==2.0.25
redis==5.0.1
pydantic==2.5.3
python-dotenv==1.0.0

# Use virtual environment
python -m venv venv
source venv/bin/activate
pip install -r requirements.txt

// package.json
{
  "dependencies": {
    "express": "^4.18.2",
    "pg": "^8.11.3",
    "redis": "^4.6.12",
    "dotenv": "^16.3.1"
  },
  "engines": {
    "node": ">=20.0.0",
    "npm": ">=10.0.0"
  }
}

// go.mod
module github.com/example/myapp

go 1.21

require (
    github.com/gin-gonic/gin v1.9.1
    github.com/lib/pq v1.10.9
    github.com/redis/go-redis/v9 v9.4.0
)

# .env.development
DATABASE_URL=postgresql://localhost:5432/myapp_dev
REDIS_URL=redis://localhost:6379/0
SECRET_KEY=dev-secret-key-not-for-production
DEBUG=true
LOG_LEVEL=DEBUG

# .env.production
DATABASE_URL=postgresql://prod-db.example.com:5432/myapp
REDIS_URL=redis://prod-redis.example.com:6379/0
SECRET_KEY=${SECRET_KEY}  # From secrets manager
DEBUG=false
LOG_LEVEL=INFO

# Backing services are swappable via config
class BackingServices:
    def __init__(self, config: Config):
        # Database - could be local PostgreSQL or AWS RDS
        self.db = create_engine(config.database_url)
        
        # Cache - could be local Redis or ElastiCache
        self.cache = redis.from_url(config.redis_url)
        
        # Email - could be SMTP or SendGrid
        self.email = EmailService(config.email_url)
        
        # Storage - could be local filesystem or S3
        self.storage = StorageService(config.storage_url)

# Swap backing services without code changes
# Development: DATABASE_URL=postgresql://localhost/dev
# Production: DATABASE_URL=postgresql://aws-rds.amazonaws.com/prod

# Build stage
docker build -t myapp:v1.2.3 .

# Release stage (combine build + config)
docker tag myapp:v1.2.3 myapp:release-456
kubectl create configmap app-config --from-env-file=.env.production

# Run stage
kubectl apply -f deployment.yaml
kubectl set image deployment/myapp myapp=myapp:release-456

# Multi-stage Dockerfile separates build and run
FROM node:20 AS builder
WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production
COPY . .
RUN npm run build

FROM node:20-slim AS runner
WORKDIR /app
COPY --from=builder /app/dist ./dist
COPY --from=builder /app/node_modules ./node_modules
CMD ["node", "dist/server.js"]

# BAD: Storing session in process memory
sessions = {}  # Lost when process restarts

@app.post("/login")
def login(username: str):
    session_id = generate_session_id()
    sessions[session_id] = {"username": username}  # Wrong!
    return {"session_id": session_id}

# GOOD: Storing session in Redis
@app.post("/login")
def login(username: str, redis: Redis = Depends(get_redis)):
    session_id = generate_session_id()
    redis.setex(
        f"session:{session_id}",
        3600,  # 1 hour TTL
        json.dumps({"username": username})
    )
    return {"session_id": session_id}

// BAD: In-memory cache
const cache = {};

app.get('/user/:id', (req, res) => {
  if (cache[req.params.id]) {
    return res.json(cache[req.params.id]);  // Wrong!
  }
  // ...
});

// GOOD: Redis cache
app.get('/user/:id', async (req, res) => {
  const cached = await redis.get(`user:${req.params.id}`);
  if (cached) {
    return res.json(JSON.parse(cached));
  }
  // ...
});

# FastAPI app binds to port
if __name__ == "__main__":
    import uvicorn
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run("main:app", host="0.0.0.0", port=port)

// Go HTTP server binds to port
func main() {
    port := os.Getenv("PORT")
    if port == "" {
        port = "8080"
    }
    
    http.HandleFunc("/", handler)
    log.Printf("Server listening on :%s", port)
    log.Fatal(http.ListenAndServe(":"+port, nil))
}

# Horizontal scaling with multiple processes
web: gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app
worker: celery -A tasks worker --concurrency=4
scheduler: celery -A tasks beat

# Kubernetes horizontal scaling
kubectl scale deployment myapp --replicas=10

# Procfile for different process types
web: uvicorn main:app --host 0.0.0.0 --port $PORT --workers 4
worker: celery -A tasks worker --loglevel=info
beat: celery -A tasks beat --loglevel=info

import signal
import sys
from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("Starting up...")
    db_pool = await create_db_pool()
    redis_client = await create_redis_client()
    
    yield  # Application runs
    
    # Shutdown
    print("Shutting down gracefully...")
    await db_pool.close()
    await redis_client.close()
    print("Shutdown complete")

app = FastAPI(lifespan=lifespan)

# Handle SIGTERM for graceful shutdown
def signal_handler(sig, frame):
    print("Received SIGTERM, shutting down...")
    sys.exit(0)

signal.signal(signal.SIGTERM, signal_handler)

// Graceful shutdown in Go
func main() {
    srv := &http.Server{Addr: ":8080", Handler: router}
    
    // Start server in goroutine
    go func() {
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("Server error: %v", err)
        }
    }()
    
    // Wait for interrupt signal
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
    <-quit
    
    log.Println("Shutting down server...")
    
    // Graceful shutdown with 30s timeout
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
    defer cancel()
    
    if err := srv.Shutdown(ctx); err != nil {
        log.Fatal("Server forced to shutdown:", err)
    }
    
    log.Println("Server exited")
}

# docker-compose.yml for local development (matches production)
version: '3.8'
services:
  app:
    build: .
    ports:
      - "8000:8000"
    environment:
      DATABASE_URL: postgresql://postgres:password@db:5432/myapp
      REDIS_URL: redis://redis:6379/0
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:16  # Same version as production
    environment:
      POSTGRES_DB: myapp
      POSTGRES_PASSWORD: password
  
  redis:
    image: redis:7-alpine  # Same version as production

import logging
import sys

# Configure logging to stdout
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[logging.StreamHandler(sys.stdout)]
)

logger = logging.getLogger(__name__)

@app.get("/users/{user_id}")
def get_user(user_id: int):
    logger.info(f"Fetching user {user_id}")
    # ...
    logger.info(f"User {user_id} fetched successfully")

// Structured logging to stdout
const winston = require('winston');

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.json(),
  transports: [
    new winston.transports.Console()
  ]
});

app.get('/users/:id', (req, res) => {
  logger.info('Fetching user', { userId: req.params.id });
  // ...
});

# Docker logs
docker logs myapp

# Kubernetes logs
kubectl logs deployment/myapp

# Aggregate with ELK, Splunk, or CloudWatch
# Logs flow: App stdout -> Container runtime -> Log aggregator

# Database migrations
kubectl run migration --image=myapp:v1.2.3 --restart=Never \
  --env="DATABASE_URL=$DATABASE_URL" \
  -- python manage.py migrate

# Django shell
kubectl run shell --image=myapp:v1.2.3 --restart=Never -it \
  --env="DATABASE_URL=$DATABASE_URL" \
  -- python manage.py shell

# One-off data fix
kubectl run data-fix --image=myapp:v1.2.3 --restart=Never \
  --env="DATABASE_URL=$DATABASE_URL" \
  -- python scripts/fix_data.py

# .env (never commit to git!)
DATABASE_URL=postgresql://user:password@localhost/db
SECRET_KEY=your-secret-key-here
API_KEY=sk-1234567890abcdef

# Load secrets from .env in development
from dotenv import load_dotenv
load_dotenv()  # Loads .env file

# Access secrets
secret_key = os.getenv("SECRET_KEY")

import boto3
import json
from functools import lru_cache

class AWSSecretsManager:
    """Fetch secrets from AWS Secrets Manager."""
    
    def __init__(self, region: str = "us-east-1"):
        self.client = boto3.client("secretsmanager", region_name=region)
    
    @lru_cache(maxsize=128)
    def get_secret(self, secret_name: str) -> dict:
        """Get secret value (cached)."""
        try:
            response = self.client.get_secret_value(SecretId=secret_name)
            return json.loads(response["SecretString"])
        except Exception as e:
            raise RuntimeError(f"Failed to fetch secret {secret_name}: {e}")
    
    def get_secret_value(self, secret_name: str, key: str) -> str:
        """Get specific key from secret."""
        secret = self.get_secret(secret_name)
        return secret.get(key)

# Usage
secrets = AWSSecretsManager()
db_password = secrets.get_secret_value("prod/database", "password")
api_key = secrets.get_secret_value("prod/api-keys", "stripe_key")

import hvac
from typing import Dict, Any

class VaultSecretsManager:
    """Fetch secrets from HashiCorp Vault."""
    
    def __init__(self, url: str, token: str):
        self.client = hvac.Client(url=url, token=token)
        if not self.client.is_authenticated():
            raise RuntimeError("Vault authentication failed")
    
    def get_secret(self, path: str) -> Dict[str, Any]:
        """Read secret from Vault."""
        response = self.client.secrets.kv.v2.read_secret_version(path=path)
        return response["data"]["data"]
    
    def set_secret(self, path: str, secret: Dict[str, Any]):
        """Write secret to Vault."""
        self.client.secrets.kv.v2.create_or_update_secret(
            path=path,
            secret=secret
        )

# Usage
vault = VaultSecretsManager(
    url="https://vault.example.com",
    token=os.getenv("VAULT_TOKEN")
)
db_creds = vault.get_secret("secret/data/database")

# Create secret from literal values
apiVersion: v1
kind: Secret
metadata:
  name: app-secrets
  namespace: production
type: Opaque
stringData:
  database-password: "super-secret-password"
  api-key: "sk-1234567890abcdef"

---
# Use secrets in deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
spec:
  template:
    spec:
      containers:
      - name: app
        image: myapp:v1.2.3
        env:
        - name: DATABASE_PASSWORD
          valueFrom:
            secretKeyRef:
              name: app-secrets
              key: database-password
        - name: API_KEY
          valueFrom:
            secretKeyRef:
              name: app-secrets
              key: api-key

# Create secret from file
kubectl create secret generic app-secrets \
  --from-file=database-password=./db-password.txt \
  --from-file=api-key=./api-key.txt

# Create secret from env file
kubectl create secret generic app-secrets \
  --from-env-file=.env.production

# ExternalSecret syncs from AWS Secrets Manager to Kubernetes
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
  name: app-secrets
  namespace: production
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: aws-secrets-manager
    kind: SecretStore
  target:
    name: app-secrets
    creationPolicy: Owner
  data:
  - secretKey: database-password
    remoteRef:
      key: prod/database
      property: password
  - secretKey: api-key
    remoteRef:
      key: prod/api-keys
      property: stripe_key

---
# SecretStore configures AWS Secrets Manager
apiVersion: external-secrets.io/v1beta1
kind: SecretStore
metadata:
  name: aws-secrets-manager
  namespace: production
spec:
  provider:
    aws:
      service: SecretsManager
      region: us-east-1
      auth:
        jwt:
          serviceAccountRef:
            name: external-secrets-sa

# ConfigMap from literal values
apiVersion: v1
kind: ConfigMap
metadata:
  name: app-config
  namespace: production
data:
  LOG_LEVEL: "INFO"
  ENVIRONMENT: "production"
  ENABLE_CACHE: "true"
  REDIS_URL: "redis://redis-service:6379"
  DATABASE_URL: "postgresql://db-service:5432/myapp"

---
# Use ConfigMap in deployment
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
spec:
  template:
    spec:
      containers:
      - name: app
        image: myapp:v1.2.3
        envFrom:
        - configMapRef:
            name: app-config
        # Or individual keys
        env:
        - name: LOG_LEVEL
          valueFrom:
            configMapKeyRef:
              name: app-config
              key: LOG_LEVEL

# Create ConfigMap from file
kubectl create configmap app-config \
  --from-file=config.yaml

# Create ConfigMap from env file
kubectl create configmap app-config \
  --from-env-file=.env.production

# Mount ConfigMap as volume
kubectl create configmap nginx-config \
  --from-file=nginx.conf

# Mount ConfigMap as file
apiVersion: v1
kind: Pod
metadata:
  name: myapp
spec:
  containers:
  - name: app
    image: myapp:v1.2.3
    volumeMounts:
    - name: config
      mountPath: /etc/config
      readOnly: true
  volumes:
  - name: config
    configMap:
      name: app-config

from pydantic import BaseSettings, Field, validator

class Settings(BaseSettings):
    """Validated configuration with Pydantic."""
    
    database_url: str = Field(..., regex=r'^postgresql://.+')
    redis_url: str = Field(..., regex=r'^redis://.+')
    secret_key: str = Field(..., min_length=32)
    port: int = Field(8000, ge=1, le=65535)
    log_level: str = Field("INFO", regex=r'^(DEBUG|INFO|WARNING|ERROR|CRITICAL)$')
    
    @validator('database_url')
    def validate_database_url(cls, v):
        if 'localhost' in v and os.getenv('ENVIRONMENT') == 'production':
            raise ValueError('Cannot use localhost database in production')
        return v
    
    class Config:
        env_file = '.env'
        case_sensitive = False

# Usage - raises ValidationError if invalid
settings = Settings()