Production Migration Runner

This document provides an example run_prod_migrations.py wrapper script. It runs Alembic commands against a production database with additional guardrails.

When to Use

• Apply already-reviewed migrations in production (upgrade head).
• Inspect current revision and history (current, history).

Avoid using production runners for creating revisions. Revisions should be created and reviewed in a shared dev/staging workflow first.

Usage

# Safe read-only commands (no confirmation required)
PYTHONPATH=. python migrations/run_prod_migrations.py current
PYTHONPATH=. python migrations/run_prod_migrations.py history

# Mutating commands (explicit confirmation required)
PYTHONPATH=. python migrations/run_prod_migrations.py upgrade head --yes
PYTHONPATH=. python migrations/run_prod_migrations.py stamp head --yes

Script Implementation

Save this file as migrations/run_prod_migrations.py.

import os
import subprocess
import sys
import logging
from dotenv import load_dotenv

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("migrations.run_prod_migrations")

load_dotenv(override=True)

APP_ENV = "production"
os.environ["ENVIRONMENT"] = APP_ENV

prod_url = os.getenv("DATABASE_URL_PROD")
if not prod_url:
    logger.error("DATABASE_URL_PROD is not set. Cannot continue.")
    sys.exit(1)

os.environ["DATABASE_URL"] = prod_url

if len(sys.argv) < 2:
    logger.info("Usage: python migrations/run_prod_migrations.py <alembic-args...> [--yes]")
    logger.info("Examples:")
    logger.info("  python migrations/run_prod_migrations.py current")
    logger.info("  python migrations/run_prod_migrations.py upgrade head --yes")
    sys.exit(1)

args = sys.argv[1:]
confirmed = False
if "--yes" in args:
    confirmed = True
    args = [a for a in args if a != "--yes"]

read_only = {"current", "history", "heads", "branches"}
command = args[0] if args else ""

if command not in read_only and not confirmed:
    logger.error("Refusing to run production mutating command without --yes. Command: %s", " ".join(args))
    sys.exit(2)

logger.info("Running Alembic in %s: alembic %s", APP_ENV, " ".join(args))
subprocess.run(["alembic", *args], check=True)