The Typical Fragile Script Anatomy

Most automation begins life like this:

# The "Danger Zone" Script
import requests
import pymysql

# 1. Hardcoded sensitive passwords in clear text
conn = pymysql.connect(host="localhost", user="root", password="password123", db="market")
# 2. Fragile synchronous requests without timeouts
resp = requests.get("https://api.external.com/v1/feed")
data = resp.json()

# 3. Direct insertion without transaction handling
with conn.cursor() as cur:
    for item in data["items"]:
        cur.execute(f"INSERT INTO data VALUES ('{item['id']}', '{item['val']}')")
conn.commit()

Why this script will inevitably wake you up at night:

  1. Security Risk: Clear-text database passwords committed to Git repositories.
  2. Socket Hangs: requests.get() without an explicit timeout=... can hang indefinitely if the remote server drops the connection during the TLS handshake.
  3. SQL Injection: String interpolation inside database queries opens dangerous injection vectors.
  4. No Retry Logic: A single dropped packet aborts the entire batch run.

Refactoring to Twelve-Factor Production Standards

Here is how that exact functionality is transformed into resilient software:

import os
import logging
from typing import List, Dict, Any
from pydantic_settings import BaseSettings
from pydantic import Field
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import sqlalchemy as sa
from sqlalchemy.orm import declarative_base, sessionmaker

# 1. Declarative Settings via Environment Variables
class ServiceSettings(BaseSettings):
    db_uri: str = Field(..., validation_alias="DATABASE_URL")
    api_endpoint: str = Field(default="https://api.external.com/v1/feed")
    api_timeout_seconds: float = 10.0

    class Config:
        env_file = ".env"

# 2. Resilient HTTP Client with Exponential Backoff
def get_resilient_session() -> requests.Session:
    session = requests.Session()
    retries = Retry(
        total=4,
        backoff_factor=1.5,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "POST"]
    )
    adapter = HTTPAdapter(max_retries=retries)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session