#!/usr/bin/env python3 """Database migration CLI tool.""" import argparse import sys from pathlib import Path from alembic import command from alembic.config import Config def main() -> None: """Main CLI function for database migrations.""" 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(): print("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) print(f"Successfully upgraded database to revision: {args.revision}") elif args.command == "downgrade": command.downgrade(alembic_cfg, args.revision) print(f"Successfully downgraded database to revision: {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) print(f"Created new auto-generated migration: {args.message}") else: command.revision(alembic_cfg, message=args.message) print(f"Created new empty migration: {args.message}") except Exception as e: print(f"Error: {e}") sys.exit(1) if __name__ == "__main__": main()