Spaces:
Paused
Paused
| import argparse | |
| import os | |
| import sys | |
| # Add current and parent directory to path to handle different execution contexts | |
| sys.path.append(os.getcwd()) | |
| sys.path.append(os.path.dirname(os.getcwd())) | |
| try: | |
| from alembic.config import Config | |
| from alembic import command | |
| from core.models import create_tables | |
| except ImportError as e: | |
| print(f"❌ critical: Failed to import required modules: {e}") | |
| sys.exit(1) | |
| def init_db(force_recreate=False): | |
| """ | |
| Initializes the production database by creating tables via SQLAlchemy | |
| and stamping the head revision for Alembic. | |
| """ | |
| print("🚀 Zenith Database Initialization Tool") | |
| print("======================================") | |
| # Check if we are using SQLite and if data directory exists | |
| from core.models.base import get_database_url | |
| db_url = get_database_url() | |
| print(f"📡 Target Database: {db_url}") | |
| if "sqlite" in db_url: | |
| db_path = db_url.replace("sqlite:///", "") | |
| if db_path.startswith("./"): | |
| db_path = os.path.join(os.getcwd(), db_path[2:]) | |
| db_dir = os.path.dirname(db_path) | |
| if db_dir and not os.path.exists(db_dir): | |
| print(f"📁 Creating directory: {db_dir}") | |
| os.makedirs(db_dir, exist_ok=True) | |
| # 1. Create tables using SQLAlchemy (Baseline) | |
| # This ensures that even if Alembic migrations are fragmented, | |
| # the latest schema defined in the models is applied. | |
| print("📦 Creating database tables from models...") | |
| try: | |
| create_tables() | |
| print("✅ Base tables created successfully.") | |
| except Exception as e: | |
| print(f"❌ Failed to create tables: {e}") | |
| # Continue anyway, as tables might already exist | |
| # 2. Stamp with Alembic | |
| # This tells Alembic that the database is already at the latest state | |
| # so it doesn't try to run 'rename' migrations on fresh tables. | |
| print("🏷️ Stamping database with Alembic HEAD...") | |
| try: | |
| alembic_cfg = Config("alembic.ini") | |
| # Ensure we point to the correct script location | |
| if not os.path.exists("alembic.ini") and os.path.exists("../alembic.ini"): | |
| alembic_cfg = Config("../alembic.ini") | |
| command.stamp(alembic_cfg, "head") | |
| print("✅ Database successfully stamped at HEAD.") | |
| except Exception as e: | |
| print(f"⚠️ Alembic stamping warning: {e}") | |
| print(" (This is normal if the database was already managed by Alembic)") | |
| print("======================================") | |
| print("🎉 Database Initialization Complete!") | |
| if __name__ == "__main__": | |
| parser = argparse.ArgumentParser(description="Initialize production database") | |
| parser.add_argument( | |
| "--force", action="store_true", help="Force recreation (not implemented)" | |
| ) | |
| args = parser.parse_args() | |
| init_db(force_recreate=args.force) | |