Production Downgrade Runner

This document provides an example run_prod_downgrade.py script to downgrade a production database with strong guardrails.

Downgrades in production are high-risk: they can permanently remove data. Prefer restoring from backup or rolling forward with corrective migrations when possible.

Usage

# Downgrade one revision (interactive confirmation)
PYTHONPATH=. python migrations/run_prod_downgrade.py -1

# Downgrade to a specific revision (interactive confirmation)
PYTHONPATH=. python migrations/run_prod_downgrade.py <revision-id>

# Skip prompt (only for emergency automation; still dangerous)
PYTHONPATH=. python migrations/run_prod_downgrade.py -1 --force

Script Implementation

Save this file as migrations/run_prod_downgrade.py.

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

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


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Safely downgrade the production database using Alembic.\n"
            "Provide either an Alembic revision (e.g. f0e1d2c3b4) or a relative step like -1.\n"
            "Use with extreme caution."
        )
    )
    parser.add_argument("target", help="Alembic downgrade target (e.g. -1, base, <revision-id>).")
    parser.add_argument("--force", action="store_true", help="Skip the interactive confirmation prompt.")
    return parser.parse_args()


def confirm(target: str) -> bool:
    logger.warning("You are about to run an Alembic downgrade against PRODUCTION.")
    logger.warning("Target revision: %s", target)
    logger.warning("This may permanently remove data. Ensure you have a recent backup.")
    response = input("Type 'CONFIRM' to proceed: ").strip()
    return response == "CONFIRM"


def main() -> None:
    args = parse_args()

    load_dotenv(override=True)
    os.environ["ENVIRONMENT"] = "production"

    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 not args.force and not confirm(args.target):
        logger.error("Confirmation failed. Aborting downgrade.")
        sys.exit(2)

    logger.info("Running: alembic downgrade %s", args.target)
    subprocess.run(["alembic", "downgrade", args.target], check=True)


if __name__ == "__main__":
    main()