Configuration Manager #

# ANTI-PATTERN: konfigurasi di kode
class Config:
    DATABASE_URL = "postgres://user:pass@localhost/mydb"
    REDIS_URL = "redis://localhost:6379"
    SECRET_KEY = "my-secret-key-123"
    DEBUG = True
    MAX_WORKERS = 4

# BENAR: konfigurasi dari environment
import os
from dataclasses import dataclass

@dataclass
class Config:
    database_url: str
    redis_url: str
    secret_key: str
    debug: bool
    max_workers: int
    log_level: str

    @classmethod
    def from_env(cls) -> 'Config':
        """Load dan validasi semua konfigurasi dari environment."""
        errors = []

        def require(name: str) -> str:
            value = os.environ.get(name)
            if not value:
                errors.append(f"Required environment variable '{name}' is not set")
                return ""
            return value

        def optional(name: str, default: str) -> str:
            return os.environ.get(name, default)

        config = cls(
            database_url=require('DATABASE_URL'),
            redis_url=optional('REDIS_URL', 'redis://localhost:6379'),
            secret_key=require('SECRET_KEY'),
            debug=optional('DEBUG', 'false').lower() == 'true',
            max_workers=int(optional('MAX_WORKERS', '4')),
            log_level=optional('LOG_LEVEL', 'INFO').upper(),
        )

        # Fail fast: jika ada yang missing, crash saat startup dengan pesan jelas
        if errors:
            raise EnvironmentError(
                "Konfigurasi tidak lengkap:\n" +
                "\n".join(f"  - {e}" for e in errors)
            )

        return config

# Load sekali saat startup
config = Config.from_env()

# .env.example — commit file ini ke git (tanpa nilai sensitif)
# Ini adalah dokumentasi konfigurasi yang dibutuhkan aplikasi

# Database (REQUIRED)
DATABASE_URL=postgres://user:password@host:5432/dbname

# Redis (REQUIRED untuk caching dan session)
REDIS_URL=redis://localhost:6379

# Security (REQUIRED)
SECRET_KEY=generate-with-openssl-rand-hex-32

# Application settings
DEBUG=false
LOG_LEVEL=INFO
MAX_WORKERS=4
PORT=8000

# External services (REQUIRED untuk fitur pembayaran)
PAYMENT_GATEWAY_URL=https://api.payment-gateway.com
PAYMENT_API_KEY=your-api-key-here

# Optional: Email service
SMTP_HOST=smtp.mailtrap.io
SMTP_PORT=587
SMTP_USER=
SMTP_PASS=

# .gitignore — PASTIKAN .env ada di sini
.env
.env.local
.env.*.local
*.secret

# .env — TIDAK di-commit ke git
# File ini ada di setiap developer machine dan server
# Dibuat dari .env.example dan diisi dengan nilai aktual

DATABASE_URL=postgres://myuser:actualpassword@localhost/mydb_dev
REDIS_URL=redis://localhost:6379
SECRET_KEY=a8f5f167f44f4964e6c998dee827110c
DEBUG=true
LOG_LEVEL=DEBUG

# Cara load .env yang benar di berbagai bahasa

# Python — menggunakan python-dotenv
from dotenv import load_dotenv
import os

# Load .env file jika ada (untuk local development)
# Di production, environment variable sudah ada dari sistem
load_dotenv()

# Node.js: require('dotenv').config()
# Go: godotenv.Load()

# Implementasi layered config sederhana

import os, json
from pathlib import Path

class LayeredConfig:
    """
    Konfigurasi dengan prioritas:
    1. Environment variable (tertinggi)
    2. config.{environment}.json
    3. config.default.json (terendah)
    """
    def __init__(self):
        self.env = os.environ.get('APP_ENV', 'development')
        self._config = {}
        self._load_file('config.default.json')
        self._load_file(f'config.{self.env}.json')
        self._load_env_vars()

    def _load_file(self, filename: str):
        path = Path(filename)
        if path.exists():
            with open(path) as f:
                self._config.update(json.load(f))

    def _load_env_vars(self):
        """Override dengan environment variable yang relevan."""
        mappings = {
            'DATABASE_URL': 'database.url',
            'REDIS_URL': 'redis.url',
            'LOG_LEVEL': 'logging.level',
            'MAX_WORKERS': 'server.max_workers',
        }
        for env_key, config_key in mappings.items():
            value = os.environ.get(env_key)
            if value:
                keys = config_key.split('.')
                d = self._config
                for k in keys[:-1]:
                    d = d.setdefault(k, {})
                d[keys[-1]] = value

    def get(self, key: str, default=None):
        """Get dengan dot notation: config.get('database.url')"""
        keys = key.split('.')
        d = self._config
        for k in keys:
            if not isinstance(d, dict) or k not in d:
                return default
            d = d[k]
        return d

# Mengambil secret dari HashiCorp Vault

import hvac, os

def get_vault_client() -> hvac.Client:
    client = hvac.Client(url=os.environ['VAULT_ADDR'])
    client.auth.approle.login(
        role_id=os.environ['VAULT_ROLE_ID'],
        secret_id=os.environ['VAULT_SECRET_ID'],
    )
    return client

def get_database_credentials() -> dict:
    """Dynamic secrets — credential baru setiap kali, dengan TTL terbatas."""
    client = get_vault_client()
    secret = client.secrets.database.generate_credentials(name='myapp-role')
    return {
        'username': secret['data']['username'],
        'password': secret['data']['password'],
        'lease_id': secret['lease_id'],
    }

def get_static_secret(path: str) -> dict:
    """Static secret dari Vault KV store."""
    client = get_vault_client()
    secret = client.secrets.kv.v2.read_secret_version(path=path)
    return secret['data']['data']

# AWS Secrets Manager — managed, tidak perlu self-host

import boto3, json, os
from functools import lru_cache

@lru_cache(maxsize=None)
def get_secret(secret_name: str) -> dict:
    """
    Ambil secret dari AWS Secrets Manager.
    Di-cache per proses — refresh dengan buat instance baru atau restart.
    """
    client = boto3.client(
        'secretsmanager',
        region_name=os.environ.get('AWS_DEFAULT_REGION', 'ap-southeast-1')
    )
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response['SecretString'])

# Penggunaan:
def get_database_config() -> dict:
    env = os.environ['APP_ENV']
    return get_secret(f"myapp/{env}/database")
    # Returns: {'host': '...', 'port': 5432, 'username': '...', 'password': '...'}

# Implementasi feature flag dengan Redis

import redis, json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

class FeatureFlag:
    CACHE_TTL = 60  # detik

    @classmethod
    def is_enabled(cls, flag_name: str, user_id: int = None) -> bool:
        flag = cls._get_flag(flag_name)
        if not flag or not flag.get('enabled'):
            return False

        # Rollout percentage: aktif untuk X% user
        rollout_pct = flag.get('rollout_percentage', 100)

        # User whitelist selalu aktif
        if user_id and user_id in flag.get('user_whitelist', []):
            return True

        # Deterministic bucket: user yang sama selalu dapat hasil yang sama
        if user_id is not None and rollout_pct < 100:
            return (user_id % 100) < rollout_pct

        return rollout_pct == 100

    @classmethod
    def _get_flag(cls, flag_name: str) -> dict | None:
        cache_key = f"feature_flag:{flag_name}"
        cached = redis_client.get(cache_key)
        if cached:
            return json.loads(cached)

        # Ambil dari database jika tidak ada di cache
        flag = FeatureFlagModel.query.filter_by(name=flag_name).first()
        if not flag:
            return None

        flag_data = {
            'enabled': flag.enabled,
            'rollout_percentage': flag.rollout_percentage,
            'user_whitelist': flag.user_whitelist or [],
        }
        redis_client.setex(cache_key, cls.CACHE_TTL, json.dumps(flag_data))
        return flag_data

# Decorator untuk endpoint
def feature_flag(flag_name: str):
    def decorator(f):
        @wraps(f)
        def decorated(*args, **kwargs):
            user_id = getattr(current_user, 'id', None)
            if not FeatureFlag.is_enabled(flag_name, user_id):
                return jsonify({'error': 'Feature not available'}), 404
            return f(*args, **kwargs)
        return decorated
    return decorator

# Penggunaan
@app.route('/api/new-checkout')
@login_required
@feature_flag('new_checkout_flow')
def new_checkout():
    return handle_new_checkout()

# Di dalam fungsi biasa
def get_recommendations(user_id: int) -> list:
    if FeatureFlag.is_enabled('ml_recommendations', user_id):
        return ml_service.get_recommendations(user_id)
    return get_popular_products()  # fallback

import os
from urllib.parse import urlparse

class ConfigValidator:
    """Validasi semua konfigurasi saat startup."""

    def __init__(self):
        self.errors = []
        self.warnings = []

    def require(self, name: str) -> 'ConfigValidator':
        if not os.environ.get(name):
            self.errors.append(f"MISSING: {name} is required")
        return self

    def require_url(self, name: str) -> 'ConfigValidator':
        value = os.environ.get(name)
        if not value:
            self.errors.append(f"MISSING: {name} is required")
        else:
            try:
                parsed = urlparse(value)
                if not parsed.scheme or not parsed.netloc:
                    self.errors.append(f"INVALID_URL: {name} = '{value}'")
            except Exception:
                self.errors.append(f"INVALID_URL: {name} = '{value}'")
        return self

    def require_int(self, name: str, min_val: int = None, max_val: int = None) -> 'ConfigValidator':
        value = os.environ.get(name)
        if not value:
            self.errors.append(f"MISSING: {name} is required")
        else:
            try:
                int_val = int(value)
                if min_val is not None and int_val < min_val:
                    self.errors.append(f"INVALID: {name} = {int_val} (min: {min_val})")
                if max_val is not None and int_val > max_val:
                    self.errors.append(f"INVALID: {name} = {int_val} (max: {max_val})")
            except ValueError:
                self.errors.append(f"NOT_INTEGER: {name} = '{value}'")
        return self

    def warn_if_default(self, name: str, dangerous_value: str) -> 'ConfigValidator':
        if os.environ.get(name) == dangerous_value:
            self.warnings.append(
                f"WARNING: {name} masih menggunakan nilai default yang tidak aman"
            )
        return self

    def validate(self):
        for warning in self.warnings:
            print(f"⚠️  {warning}")
        if self.errors:
            error_msg = "\n".join(f"  ✗ {e}" for e in self.errors)
            raise SystemExit(
                f"\n❌ Konfigurasi tidak valid:\n\n{error_msg}\n\n"
                f"Lihat .env.example untuk panduan.\n"
            )
        print(f"✓ Konfigurasi valid")

# Panggil di awal aplikasi, sebelum apapun
def validate_config():
    ConfigValidator() \
        .require_url('DATABASE_URL') \
        .require_url('REDIS_URL') \
        .require('SECRET_KEY') \
        .require_int('MAX_WORKERS', min_val=1, max_val=64) \
        .require_int('PORT', min_val=1024, max_val=65535) \
        .warn_if_default('SECRET_KEY', 'changeme') \
        .warn_if_default('SECRET_KEY', 'your-secret-key-here') \
        .validate()

validate_config()
app = create_app()

# ✗ Anti-pattern 1: secret di source code
API_KEY = "sk-prod-xxxxxxxxxxxxxxxx"
DATABASE_PASSWORD = "supersecretpassword"
# ✓ Solusi: environment variable + secrets manager

# ✗ Anti-pattern 2: .env di-commit ke git
# .gitignore tidak mengandung .env
# ✓ Solusi: .env di .gitignore DARI AWAL, .env.example yang di-commit

# ✗ Anti-pattern 3: tidak ada validasi konfigurasi
# Aplikasi start tanpa DATABASE_URL → crash saat ada request
# ✓ Solusi: validasi dan fail fast saat startup dengan pesan jelas

# ✗ Anti-pattern 4: .env.example tidak up-to-date
# Production punya 10 env var yang tidak ada di .env.example
# Developer baru debugging 2 jam
# ✓ Solusi: .env.example selalu di-update bersamaan dengan perubahan konfigurasi

# ✗ Anti-pattern 5: feature flag tanpa mekanisme cleanup
# Flag yang sudah 100% rollout sejak 6 bulan tidak pernah dihapus
# Kode lama dan kondisional yang tidak perlu terus ada
# ✓ Solusi: feature flag punya expiry date, cleanup secara rutin

# ✗ Anti-pattern 6: cache secret terlalu lama
# Secret diambil saat startup, di-cache selamanya
# Secret rotation di Vault tidak berefek
# ✓ Solusi: cache secret dengan TTL pendek (< 1 jam)