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:
- Security Risk: Clear-text database passwords committed to Git repositories.
- Socket Hangs:
requests.get()without an explicittimeout=...can hang indefinitely if the remote server drops the connection during the TLS handshake. - SQL Injection: String interpolation inside database queries opens dangerous injection vectors.
- 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