My Scripting Best Practices: A Production-Ready Checklist

Lessons learned from my years of writing data migration, backfill, and operational scripts.

The Problem

We've all been there. You write a "quick script" to fix some data, run it in production, and then spend the next 4 hours dealing with:

• Unexpected timeouts
• Memory blowups
• Partial updates with no rollback
• No idea how long it will take
• No visibility into progress

After enough painful incidents, I developed a rigorous checklist. Here are my hard-won best practices.

1. Standardized Template Table

Every script I write starts with this exact header:

2. Always Provide an Estimated Total Runtime

Before the script even runs, the user should know:

Estimated total runtime: ~45 minutes (based on 2.1M records @ ~1,200 records/sec)

How to calculate this:

• Sample 10,000 records and measure per-record time
• Multiply by total count
• Add 20% buffer for variance

Example:

def estimate_runtime(sample_size=10000):
    start = time.time()
    process_sample(sample_size)
    elapsed = time.time() - start
    per_record = elapsed / sample_size
    total_est = per_record * total_count
    print(f"Estimated: {total_est/60:.1f} minutes")
    return total_est

3. Runtime Evaluation Protocol

Don't just guess. Build a progressive estimation system:

class RuntimeEstimator:
    def __init__(self, total):
        self.total = total
        self.processed = 0
        self.start = time.time()
        self.window = []

    def update(self, batch_size):
        self.processed += batch_size
        self.window.append(time.time())
        if len(self.window) > 10:
            self.window.pop(0)

        if self.processed >= 100:
            elapsed = self.window[-1] - self.window[0]
            rate = self.processed / elapsed
            remaining = (self.total - self.processed) / rate
            eta = time.time() + remaining
            print(f"ETA: {eta.strftime('%H:%M:%S')} | Rate: {rate:.0f} rec/s")

4. Comprehensive Logging Output

Every script must produce:

▸ a) Expected Target Count

[TARGET] Total records to process: 2,134,567

▸ b) Success/Failure Counts

[PROGRESS] Processed: 1,200,000 | Success: 1,198,234 | Failed: 1,766

▸ c) Real-time Progress (with percentage)

[PROGRESS] 56.2% | 1,200,000 / 2,134,567 | ETA: 14:32:15

▸ Implementation:

LOG_INTERVAL = 1000  # update every 1000 records

def log_progress(processed, success, failed, total, start_time):
    pct = (processed / total) * 100
    rate = processed / (time.time() - start_time)
    eta_sec = (total - processed) / rate if rate > 0 else 0

    print(json.dumps({
        "event": "progress",
        "percentage": round(pct, 2),
        "processed": processed,
        "success": success,
        "failed": failed,
        "rate": round(rate, 1),
        "eta_seconds": round(eta_sec),
        "eta": (datetime.now() + timedelta(seconds=eta_sec)).isoformat()
    }))

5. Data Modification = Backup + Rollback

If your script writes, updates, or deletes anything, you must provide:

▸ Backup Strategy:

-- Pre-execution backup
CREATE TABLE users_backup_20260627 AS SELECT * FROM users;

-- Or for large tables, use pg_dump
pg_dump -t users > /backups/users_pre_script_20260627.sql

▸ Rollback Script:

-- Rollback: revert to backup
TRUNCATE users;
INSERT INTO users SELECT * FROM users_backup_20260627;

-- Or if only certain rows were affected
UPDATE users u
SET score = b.score
FROM users_backup_20260627 b
WHERE u.id = b.id AND u.updated_at >= '2026-06-27 10:00:00';

▸ In the script itself:

def main():
    if not args.dry_run and not confirm_destructive():
        sys.exit(0)

    if args.backup:
        execute_backup()

    try:
        run_migration()
    except Exception as e:
        if args.auto_rollback:
            rollback()
        raise

6. Idempotency — Always Specify

Every script must clearly state whether it supports idempotent execution:

✅ Idempotent: Can be run multiple times. Uses INSERT ... ON CONFLICT DO NOTHING / UPDATE ... WHERE changed = false

❌ Not idempotent: Appends to a log table without deduplication. Do not rerun.

How to implement idempotency:

# Use a checkpoint table
CREATE TABLE script_checkpoints (
    script_name TEXT,
    last_processed_id BIGINT,
    run_id UUID,
    PRIMARY KEY (script_name, run_id)
);

# Store progress to resume
def process_batch(start_id):
    # Only process records with id > last_processed_id
    pass

7. Flexibility — Make It Configurable

Never hardcode:

• Database connection strings
• Table names
• Batch sizes
• Timeouts
• Retry counts

Use environment variables + CLI args:

# config.py
BATCH_SIZE = int(os.getenv("BATCH_SIZE", 1000))
TIMEOUT = int(os.getenv("TIMEOUT_SECONDS", 30))
MAX_RETRIES = int(os.getenv("MAX_RETRIES", 3))
CONCURRENCY = int(os.getenv("CONCURRENCY", 4))

# CLI overrides
parser.add_argument("--batch-size", type=int)
parser.add_argument("--limit", type=int)  # for testing
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--from-date", type=date)

8. Handling Interruptions (SIGINT, SIGTERM)

Scripts must handle clean shutdown:

import signal
import sys

class GracefulKiller:
    kill_now = False

    def __init__(self):
        signal.signal(signal.SIGINT, self.exit_gracefully)
        signal.signal(signal.SIGTERM, self.exit_gracefully)

    def exit_gracefully(self, signum, frame):
        self.kill_now = True

def main():
    killer = GracefulKiller()
    checkpoint = load_checkpoint()

    for batch in batches:
        if killer.kill_now:
            print("\n[SHUTDOWN] Received interrupt, saving checkpoint...")
            save_checkpoint(current_id)
            print("[SHUTDOWN] Checkpoint saved. Run with --resume to continue.")
            sys.exit(0)
        process_batch(batch)

Always log where you stopped:

[CHECKPOINT] Saved at ID: 1,234,567. Run with --resume to continue.

9. Estimated Execution Duration (Pre-run)

Provide multiple estimates:

Estimated runtime:
- Best case: 30 min (assuming 2,000 rec/s)
- Expected: 45 min (1,200 rec/s)
- Worst case: 2 hrs (500 rec/s with retries)

10. Execution Log Output

All scripts must write structured logs:

import logging
logging.basicConfig(
    filename=f'/var/log/scripts/{script_name}_{date}.log',
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)

# Also output to stdout for real-time monitoring
console = logging.StreamHandler()
console.setLevel(logging.INFO)
logging.getLogger().addHandler(console)

# Structured logging (JSON for easy parsing)
logger.info(json.dumps({
    "event": "batch_complete",
    "batch": batch_num,
    "records": len(batch),
    "duration_sec": elapsed
}))

11. Data Backup Assessment

Before running, answer:

12. Large-Scale Production Assessment

For scripts affecting >1M records, run a pre-flight assessment:

-- Check table size
SELECT pg_size_pretty(pg_total_relation_size('users'));

-- Check distribution
SELECT COUNT(*), MIN(created_at), MAX(created_at) FROM users;

-- Check index usage (will your query use indexes?)
EXPLAIN ANALYZE SELECT * FROM users WHERE updated_at > '2026-01-01';

Document the results:

Production Assessment:
- Total records: 14.2M
- Affected records: 2.1M (15%)
- Table size: 4.2GB
- Index available: idx_users_updated_at (yes)
- Expected I/O: ~500MB read, ~200MB write
- Database CPU impact: ~15% during run

13. Memory Usage Testing

Always test memory before production:

# Test with 1/10 of production data
python script.py --limit 200000 --batch-size 5000

# Monitor memory
/usr/bin/time -v python script.py --limit 200000
# or
memory-profiler python script.py

# In production, set memory limits
ulimit -v 1048576  # 1GB virtual memory limit

Implementation:

import tracemalloc

tracemalloc.start()

# ... after heavy operation ...
current, peak = tracemalloc.get_traced_memory()
print(f"Memory: {current / 1024 / 1024:.2f} MB (peak: {peak / 1024 / 1024:.2f} MB)")

14. Run Short Scripts First

If you have multiple scripts to run:

# Sort by estimated duration (shortest first)
./scripts/backfill_small_table.py  # 2 min
./scripts/update_user_metadata.py  # 15 min
./scripts/migrate_audit_logs.py    # 2 hours

Why? Early failures are detected faster, and rollback is easier.

15. Real-Time Statistics Scripts Must Include Historical Fix Scripts

This is non-negotiable:

🚨 If you're deploying a script that computes real-time stats (e.g., daily active users, revenue aggregates), you MUST provide a companion historical backfill script.

Why? When the real-time script has a bug, you need to recompute historical data.

Example:

# realtime_user_stats.py — runs every 5 min, updates last 24h
# historical_user_stats.py — backfills from a start date
#   python historical_user_stats.py --from-date 2026-01-01

Review checklist:

• Real-time script deployed
• Historical script reviewed and tested
• Historical script supports --from-date and --to-date
• Historical script is idempotent
• Historical script can run without affecting production

16. Special Exception Handling (Redis, Cache, External Dependencies)

Worst-case scenario: Redis gets flushed mid-script.

▸ Pre-mortem analysis:

▸ Redis failure response steps:

1. Script detects Redis connection error
2. Switch to "degraded mode" — read directly from DB
3. Log all cache misses with Redis error
4. After script completes, rebuild Redis cache:
   python rebuild_cache.py --start-id 0 --batch-size 10000
5. Verify cache consistency with DB
6. If Redis is unrecoverable, run full cache rebuild

Code:

def get_with_cache_fallback(key, db_query):
    try:
        val = redis_client.get(key)
        if val is not None:
            return val
    except redis.ConnectionError as e:
        logger.error(f"Redis unavailable: {e}. Falling back to DB.")

    # Fallback to DB
    val = db_query()

    # Try to cache it (if Redis recovers)
    try:
        redis_client.setex(key, 3600, val)
    except:
        pass  # Redis still down, ignore

    return val

Final Checklist Before Any Production Script

• Standardized header table completed
• Runtime estimated and displayed
• Progress logging implemented (target, success, fail, ETA)
• Dry-run mode available (--dry-run)
• Backup strategy documented and tested
• Rollback script exists (for write scripts)
• Idempotency explicitly stated
• Graceful shutdown (SIGINT/SIGTERM) implemented
• Memory usage tested with sample data
• Configurable via env vars + CLI
• Historical fix script exists (for real-time stats scripts)
• Redis/DB failure scenarios documented
• Logs are structured (JSON) and written to file + stdout
• Code reviewed by at least one peer
• Run on staging with production-like data volume

Conclusion

These aren't just "nice-to-haves." In production, they're survival requirements. A well-written script:

• Saves you from 3 AM wake-up calls
• Makes rollback a 30-second operation, not a crisis
• Gives stakeholders confidence
• Allows juniors to run scripts safely

The extra 20% effort in preparation saves 200% in incident remediation.

What are your must-have script practices? Share them in the comments!

Follow me for more production engineering deep dives.