150 lines
4.4 KiB
Python
150 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Migration script to add site_pages table for Story 3.4
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
project_root = Path(__file__).parent.parent
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from sqlalchemy import text, inspect
|
|
from src.database.session import db_manager
|
|
from src.core.config import get_config
|
|
|
|
|
|
def check_table_exists(inspector, table_name: str) -> bool:
|
|
"""Check if a table exists"""
|
|
return table_name in inspector.get_table_names()
|
|
|
|
|
|
def migrate_add_site_pages_table(connection):
|
|
"""Create site_pages table"""
|
|
print("\n[1/1] Creating site_pages table...")
|
|
|
|
inspector = inspect(connection)
|
|
|
|
if check_table_exists(inspector, 'site_pages'):
|
|
print(" [INFO] Table site_pages already exists, skipping")
|
|
return True
|
|
|
|
try:
|
|
connection.execute(text("""
|
|
CREATE TABLE site_pages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
site_deployment_id INTEGER NOT NULL,
|
|
page_type VARCHAR(20) NOT NULL,
|
|
content TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
|
|
FOREIGN KEY (site_deployment_id) REFERENCES site_deployments(id) ON DELETE CASCADE,
|
|
UNIQUE (site_deployment_id, page_type)
|
|
)
|
|
"""))
|
|
print(" [OK] Created site_pages table")
|
|
|
|
connection.execute(text("""
|
|
CREATE INDEX idx_site_pages_site ON site_pages(site_deployment_id)
|
|
"""))
|
|
print(" [OK] Created index on site_deployment_id")
|
|
|
|
connection.execute(text("""
|
|
CREATE INDEX idx_site_pages_type ON site_pages(page_type)
|
|
"""))
|
|
print(" [OK] Created index on page_type")
|
|
|
|
connection.commit()
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f" [ERROR] {e}")
|
|
connection.rollback()
|
|
return False
|
|
|
|
|
|
def verify_migration(connection):
|
|
"""Verify the migration was successful"""
|
|
print("\n[Verification] Checking migration results...")
|
|
|
|
inspector = inspect(connection)
|
|
|
|
success = True
|
|
|
|
if check_table_exists(inspector, 'site_pages'):
|
|
print(" [OK] site_pages table exists")
|
|
|
|
columns = inspector.get_columns('site_pages')
|
|
expected_columns = ['id', 'site_deployment_id', 'page_type', 'content', 'created_at', 'updated_at']
|
|
actual_columns = [col['name'] for col in columns]
|
|
|
|
for col in expected_columns:
|
|
if col in actual_columns:
|
|
print(f" [OK] site_pages.{col} exists")
|
|
else:
|
|
print(f" [ERROR] site_pages.{col} MISSING")
|
|
success = False
|
|
|
|
indexes = inspector.get_indexes('site_pages')
|
|
index_names = [idx['name'] for idx in indexes]
|
|
print(f" [INFO] Indexes: {index_names}")
|
|
else:
|
|
print(" [ERROR] site_pages table MISSING")
|
|
success = False
|
|
|
|
return success
|
|
|
|
|
|
def main():
|
|
"""Run the migration"""
|
|
print("=" * 60)
|
|
print("Story 3.4 Database Migration - Site Pages")
|
|
print("=" * 60)
|
|
|
|
try:
|
|
config = get_config()
|
|
print(f"\nDatabase: {config.database.url}")
|
|
except Exception as e:
|
|
print(f"[ERROR] Error loading configuration: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
db_manager.initialize()
|
|
engine = db_manager.get_engine()
|
|
connection = engine.connect()
|
|
except Exception as e:
|
|
print(f"[ERROR] Error connecting to database: {e}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
success = migrate_add_site_pages_table(connection)
|
|
|
|
if success:
|
|
if verify_migration(connection):
|
|
print("\n" + "=" * 60)
|
|
print("[SUCCESS] Migration completed successfully!")
|
|
print("=" * 60)
|
|
else:
|
|
print("\n" + "=" * 60)
|
|
print("[WARNING] Migration completed with warnings")
|
|
print("=" * 60)
|
|
else:
|
|
print("\n" + "=" * 60)
|
|
print("[FAILED] Migration failed!")
|
|
print("=" * 60)
|
|
sys.exit(1)
|
|
|
|
except Exception as e:
|
|
print(f"\n[ERROR] Unexpected error during migration: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|
|
finally:
|
|
connection.close()
|
|
db_manager.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|