101 lines
3.0 KiB
Python
Executable File
101 lines
3.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Database migration CLI tool."""
|
|
|
|
import argparse
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from alembic.config import Config
|
|
|
|
from alembic import command
|
|
|
|
# Set up logging
|
|
logger = logging.getLogger(__name__)
|
|
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
|
|
|
|
|
def main() -> None:
|
|
"""Run database migration CLI tool."""
|
|
parser = argparse.ArgumentParser(description="Database migration tool")
|
|
subparsers = parser.add_subparsers(dest="command", help="Migration commands")
|
|
|
|
# Upgrade command
|
|
upgrade_parser = subparsers.add_parser(
|
|
"upgrade", help="Upgrade database to latest revision",
|
|
)
|
|
upgrade_parser.add_argument(
|
|
"revision",
|
|
nargs="?",
|
|
default="head",
|
|
help="Target revision (default: head)",
|
|
)
|
|
|
|
# Downgrade command
|
|
downgrade_parser = subparsers.add_parser("downgrade", help="Downgrade database")
|
|
downgrade_parser.add_argument("revision", help="Target revision")
|
|
|
|
# Current command
|
|
subparsers.add_parser("current", help="Show current revision")
|
|
|
|
# History command
|
|
subparsers.add_parser("history", help="Show revision history")
|
|
|
|
# Generate migration command
|
|
revision_parser = subparsers.add_parser("revision", help="Create new migration")
|
|
revision_parser.add_argument(
|
|
"-m", "--message", required=True, help="Migration message",
|
|
)
|
|
revision_parser.add_argument(
|
|
"--autogenerate", action="store_true", help="Auto-generate migration",
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if not args.command:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
# Get the alembic config
|
|
config_path = Path("alembic.ini")
|
|
if not config_path.exists():
|
|
logger.error("Error: alembic.ini not found. Run from the backend directory.")
|
|
sys.exit(1)
|
|
|
|
alembic_cfg = Config(str(config_path))
|
|
|
|
try:
|
|
if args.command == "upgrade":
|
|
command.upgrade(alembic_cfg, args.revision)
|
|
logger.info(
|
|
"Successfully upgraded database to revision: %s", args.revision,
|
|
)
|
|
|
|
elif args.command == "downgrade":
|
|
command.downgrade(alembic_cfg, args.revision)
|
|
logger.info(
|
|
"Successfully downgraded database to revision: %s", args.revision,
|
|
)
|
|
|
|
elif args.command == "current":
|
|
command.current(alembic_cfg)
|
|
|
|
elif args.command == "history":
|
|
command.history(alembic_cfg)
|
|
|
|
elif args.command == "revision":
|
|
if args.autogenerate:
|
|
command.revision(alembic_cfg, message=args.message, autogenerate=True)
|
|
logger.info("Created new auto-generated migration: %s", args.message)
|
|
else:
|
|
command.revision(alembic_cfg, message=args.message)
|
|
logger.info("Created new empty migration: %s", args.message)
|
|
|
|
except (OSError, RuntimeError):
|
|
logger.exception("Error occurred during migration")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|