Spaces:
Paused
Paused
File size: 2,875 Bytes
4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac 4ae946d 4e400ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | 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)
|