Twelve-Factor App Methodology Explained — Build Software That Scales

# ============================================================
# Demonstrating Factors I-IV in a real Docker Compose setup
# This file shows how a twelve-factor app wires its foundation
# ============================================================

version: '3.9'

services:

  # --- The Application (Factor I: one codebase, one image) ---
  web_api:
    build:
      context: .          # Build from THIS repo — one codebase, one image
      dockerfile: Dockerfile
    ports:
      - "8000:8000"

    # Factor III: ALL config via environment variables
    # Nothing here is hardcoded in application source code
    environment:
      - APP_ENV=development
      - SECRET_KEY=dev-only-secret-replace-in-prod  # In prod, inject via secrets manager
      - LOG_LEVEL=debug

      # Factor IV: Backing services treated as attached resources via URL
      # Swap DATABASE_URL to point at RDS in prod — zero code changes needed
      - DATABASE_URL=postgresql://app_user:app_pass@postgres_db:5432/appdb
      - CACHE_URL=redis://cache_store:6379/0
      - EMAIL_API_URL=https://api.sendgrid.com/v3/mail/send
      - EMAIL_API_KEY=SG.placeholder-replace-with-real-key

    depends_on:
      - postgres_db
      - cache_store

  # --- Postgres (a Backing Service — Factor IV) ---
  postgres_db:
    image: postgres:15-alpine
    environment:
      - POSTGRES_USER=app_user
      - POSTGRES_PASSWORD=app_pass
      - POSTGRES_DB=appdb
    volumes:
      - postgres_data:/var/lib/postgresql/data  # Persist data outside container

  # --- Redis Cache (another Backing Service — Factor IV) ---
  cache_store:
    image: redis:7-alpine

volumes:
  postgres_data:

# ============================================================
# Factor II: Explicit dependencies are in requirements.txt
# (shown below — never pip install globally without pinning)
# ============================================================

# requirements.txt (referenced by Dockerfile)
# fastapi==0.110.0
# uvicorn==0.29.0
# psycopg2-binary==2.9.9
# redis==5.0.3
# httpx==0.27.0
# python-dotenv==1.0.1   # Only for local dev — prod uses real env vars

# ============================================================
# Factor VI: Stateless Processes — the right way
# This FastAPI app stores ALL session state in Redis,
# so ANY running instance can handle ANY request.
# Scale to 50 instances — every one works identically.
# ============================================================

import os
import json
import uuid
from fastapi import FastAPI, HTTPException, Cookie
from fastapi.responses import JSONResponse
import redis

app = FastAPI()

# Factor IV + III: Backing service URL from environment variable
SESSION_STORE = redis.from_url(
    os.environ["CACHE_URL"],  # e.g. redis://cache_store:6379/0
    decode_responses=True
)

SESSION_TTL_SECONDS = 3600  # Sessions expire after 1 hour


@app.post("/login")
async def login(username: str, password: str):
    """
    Factor VI in action: we create a session token and store
    ALL session data in Redis — nothing lives in process memory.
    Any instance of this app can validate this session.
    """
    # (In reality, verify username/password against database)
    if password != "correct-horse-battery-staple":
        raise HTTPException(status_code=401, detail="Invalid credentials")

    # Generate a unique session ID
    session_id = str(uuid.uuid4())

    # Store session data in Redis (the backing service) — NOT in process memory
    session_data = {
        "username": username,
        "role": "editor",
        "login_timestamp": "2024-01-15T09:30:00Z"
    }
    SESSION_STORE.setex(
        name=f"session:{session_id}",   # Namespaced key
        time=SESSION_TTL_SECONDS,         # Auto-expire old sessions
        value=json.dumps(session_data)    # Serialised to string for Redis
    )

    response = JSONResponse({"message": "Login successful", "session_id": session_id})
    # Return session_id to client via cookie
    response.set_cookie(key="session_id", value=session_id, httponly=True)
    return response


@app.get("/dashboard")
async def dashboard(session_id: str = Cookie(default=None)):
    """
    Any of the 50 running instances can serve this request
    because session state lives in Redis, not in process memory.
    This is what makes horizontal scaling work.
    """
    if not session_id:
        raise HTTPException(status_code=401, detail="No session cookie")

    # Look up session from Redis — works regardless of which instance handles this
    raw_session = SESSION_STORE.get(f"session:{session_id}")

    if not raw_session:
        raise HTTPException(status_code=401, detail="Session expired or invalid")

    session_data = json.loads(raw_session)

    return {
        "welcome": f"Hello, {session_data['username']}!",
        "role": session_data["role"],
        "instance_note": "Any instance served this — stateless processes working correctly"
    }


# ============================================================
# Factor VII: Port Binding — app is self-contained
# Run with: uvicorn stateless_web_process:app --host 0.0.0.0 --port 8000
# No Apache/Nginx dependency at runtime. The app IS the server.
# ============================================================

# ============================================================
# Factor VIII: Concurrency — Procfile for process type hierarchy
# web: uvicorn stateless_web_process:app --host 0.0.0.0 --port $PORT --workers 4
# worker: celery -A tasks worker --concurrency=8
# scheduler: celery -A tasks beat
# Scale web and worker independently — no single giant process
# ============================================================

# ============================================================
# Factor IX: Disposability — graceful shutdown
# Factor XI: Logs as event streams — write to stdout only
# ============================================================

import os
import sys
import signal
import logging
import time
from threading import Event

# Factor XI: Configure logging to stdout ONLY.
# The platform (Heroku/K8s/ECS) captures this and routes it.
# Your app NEVER writes to /var/log/app.log or manages log rotation.
logging.basicConfig(
    stream=sys.stdout,              # stdout only — no file handlers
    level=logging.INFO,
    format='{"timestamp": "%(asctime)s", "level": "%(levelname)s", "message": "%(message)s"}'
    # Structured JSON logs are even better — tools like Datadog parse these automatically
)

logger = logging.getLogger(__name__)


class OrderProcessingWorker:
    """
    A background worker that processes orders from a queue.
    Demonstrates Factor IX: fast startup + graceful SIGTERM handling.
    """

    def __init__(self):
        self.is_running = False
        self.shutdown_event = Event()
        self.current_job_id = None

        # Register signal handlers for graceful shutdown
        # SIGTERM is sent by Heroku/Kubernetes when scaling down or deploying
        signal.signal(signal.SIGTERM, self._handle_shutdown_signal)
        signal.signal(signal.SIGINT, self._handle_shutdown_signal)   # Ctrl+C locally

    def _handle_shutdown_signal(self, signum, frame):
        """
        Factor IX: When the platform sends SIGTERM, we don't die immediately.
        We finish the current job, return incomplete work to the queue,
        then exit cleanly. Zero data loss.
        """
        signal_name = "SIGTERM" if signum == 15 else "SIGINT"
        logger.info(f"Received {signal_name} — starting graceful shutdown")

        if self.current_job_id:
            logger.info(f"Returning job {self.current_job_id} to queue before shutdown")
            # In a real app: queue.nack(self.current_job_id)  — returns job to queue
            # so another worker picks it up. No lost orders.

        self.shutdown_event.set()  # Signal the main loop to stop

    def process_order(self, order_id: str, order_data: dict) -> bool:
        """Simulate processing a single order."""
        self.current_job_id = order_id
        logger.info(f"Processing order {order_id} for customer {order_data['customer_email']}")

        # Simulate work (database writes, payment processing, etc.)
        time.sleep(2)

        logger.info(f"Order {order_id} processed successfully — total: ${order_data['total_cents'] / 100:.2f}")
        self.current_job_id = None
        return True

    def run(self):
        """Main processing loop."""
        self.is_running = True
        logger.info("Order worker started — listening for jobs")  # Logs to stdout

        # Simulate picking up jobs from a queue
        pending_orders = [
            {"id": "ord_001", "customer_email": "alice@example.com", "total_cents": 4999},
            {"id": "ord_002", "customer_email": "bob@example.com",   "total_cents": 12500},
            {"id": "ord_003", "customer_email": "carol@example.com", "total_cents": 899},
        ]

        for order in pending_orders:
            if self.shutdown_event.is_set():
                logger.info("Shutdown requested — stopping before next job")
                break  # Clean exit — don't start a new job if shutting down

            self.process_order(order["id"], order)

        logger.info("Worker shut down cleanly")  # This reaches your log aggregator
        sys.exit(0)  # Clean exit code — platform knows this was intentional


# ============================================================
# Factor XII: Admin process example
# Run database migrations as a one-off process:
#   heroku run python manage.py db upgrade
# or in Docker:
#   docker run --env-file .env myapp:v1.2 python manage.py db upgrade
# Same image, same config, same codebase as production — guaranteed consistency
# ============================================================

if __name__ == "__main__":
    worker = OrderProcessingWorker()
    worker.run()