88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Database initialization script that creates all tables and triggers.
|
|
Use this for setting up fresh development or production databases.
|
|
|
|
Usage:
|
|
python database_init.py
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
from sqlalchemy import create_engine, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
# Add parent directory to path to import models
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from models import Base
|
|
from database import get_database_url
|
|
|
|
def init_database():
|
|
"""Initialize database with all tables and triggers"""
|
|
|
|
database_url = get_database_url()
|
|
engine = create_engine(database_url)
|
|
|
|
print("🚀 Initializing database...")
|
|
print(f"📍 Database URL: {database_url}")
|
|
|
|
try:
|
|
# Create all tables first
|
|
print("📊 Creating tables...")
|
|
Base.metadata.create_all(bind=engine)
|
|
print("✅ Tables created successfully")
|
|
|
|
# Verify critical tables exist before creating triggers
|
|
print("🔍 Verifying tables exist...")
|
|
with engine.connect() as connection:
|
|
# Check if products and products_history tables exist
|
|
products_exists = connection.execute(text("""
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM information_schema.tables
|
|
WHERE table_name = 'products'
|
|
);
|
|
""")).scalar()
|
|
|
|
history_exists = connection.execute(text("""
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM information_schema.tables
|
|
WHERE table_name = 'products_history'
|
|
);
|
|
""")).scalar()
|
|
|
|
if not products_exists:
|
|
raise Exception("Products table was not created")
|
|
if not history_exists:
|
|
raise Exception("Products history table was not created")
|
|
|
|
print("✅ Required tables verified")
|
|
|
|
# Create triggers (if not already created by event listener)
|
|
print("⚙️ Ensuring triggers are created...")
|
|
with engine.connect() as connection:
|
|
# Check if trigger exists
|
|
result = connection.execute(text("""
|
|
SELECT EXISTS (
|
|
SELECT 1 FROM information_schema.triggers
|
|
WHERE trigger_name = 'products_versioning_trigger'
|
|
);
|
|
""")).scalar()
|
|
|
|
if not result:
|
|
print("📝 Creating products versioning trigger...")
|
|
from models import PRODUCTS_VERSIONING_TRIGGER_SQL
|
|
connection.execute(text(PRODUCTS_VERSIONING_TRIGGER_SQL))
|
|
connection.commit()
|
|
print("✅ Trigger created successfully")
|
|
else:
|
|
print("✅ Trigger already exists")
|
|
|
|
print("🎉 Database initialization completed successfully!")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error initializing database: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
init_database() |