flowCreate.solutions

Development Downgrade Runner

This document provides an example run_dev_downgrade.py script to downgrade a development database with explicit confirmation.

Downgrades are destructive: they can drop tables/columns or remove data. Use with caution.

Usage

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

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

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

Script Implementation

Save this file as migrations/run_dev_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_dev_downgrade")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Downgrade the development database using Alembic.\n"
            "Provide an Alembic revision (e.g. f0e1d2c3b4) or a relative step like -1."
        )
    )
    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 downgrade the DEVELOPMENT database.")
    logger.warning("Target revision: %s", target)
    logger.warning("This may drop or modify schema/data. Make sure you have backups if needed.")
    response = input("Type 'CONFIRM' to proceed: ").strip()
    return response == "CONFIRM"


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

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

    dev_url = os.getenv("DATABASE_URL_DEV")
    if not dev_url:
        logger.error("DATABASE_URL_DEV is not set. Cannot continue.")
        sys.exit(1)
    os.environ["DATABASE_URL"] = dev_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()