Spaces:
Sleeping
Sleeping
File size: 2,712 Bytes
01d5a5d |
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 83 84 85 86 87 88 89 |
#!/usr/bin/env python
"""
Database Migration Runner
This script runs database migrations using the migration manager.
It should be executed whenever the database schema needs to be updated.
"""
import os
import sys
from pathlib import Path
# Add project root to path
project_root = str(Path(__file__).parent.parent)
sys.path.insert(0, project_root)
from lpm_kernel.configs.config import Config
from lpm_kernel.database.migration_manager import MigrationManager
from lpm_kernel.common.logging import logger
def get_db_path():
"""Get the database path from environment or use default"""
config = Config.from_env()
db_path = config.get("SQLITE_DB_PATH", os.path.join(project_root, "data", "sqlite", "lpm.db"))
return db_path
def run_migrations():
"""Run all pending database migrations"""
db_path = get_db_path()
# logger.info(f"Using database at: {db_path}")
# Check if database file exists
if not os.path.exists(db_path):
# logger.error(f"Database file not found at {db_path}")
return False
try:
# Initialize migration manager
migrations_dir = os.path.join(project_root, "lpm_kernel", "database", "migrations")
manager = MigrationManager(db_path)
# Apply migrations
applied = manager.apply_migrations(migrations_dir)
# if applied:
# logger.info(f"Successfully applied {len(applied)} migrations")
# else:
# logger.info("No new migrations to apply")
return True
except Exception as e:
logger.error(f"Error during migrations: {e}")
return False
def create_migration(description):
"""Create a new migration file"""
db_path = get_db_path()
migrations_dir = os.path.join(project_root, "lpm_kernel", "database", "migrations")
manager = MigrationManager(db_path)
filepath = manager.create_migration(description, migrations_dir)
# logger.info(f"Created new migration at: {filepath}")
return filepath
if __name__ == "__main__":
# logger.info("Starting database migration")
if len(sys.argv) > 1 and sys.argv[1] == "create":
if len(sys.argv) > 2:
description = sys.argv[2]
create_migration(description)
else:
logger.error("Missing migration description")
print("Usage: python run_migrations.py create 'Add new table'")
sys.exit(1)
else:
success = run_migrations()
if success:
# logger.info("Migration completed successfully")
sys.exit(0)
else:
logger.error("Migration failed")
sys.exit(1)
|