Add helper functions for Migration

helpers/migration_helpers.py

@@ -0,0 +1,97 @@
+from typing import Optional, List, Dict, Union
+from datetime import datetime
+import hashlib
+import os
+from sqlalchemy.orm import Session
+from sqlalchemy import desc
+from models.migration import Migration
+
+def calculate_script_checksum(script_path: str) -> str:
+    """
+    Calculate SHA-256 checksum of migration script file.
+    
+    Args:
+        script_path: Path to the migration script file
+        
+    Returns:
+        str: Hexadecimal checksum of the file
+    """
+    if not os.path.exists(script_path):
+        raise FileNotFoundError(f"Migration script not found at: {script_path}")
+        
+    sha256_hash = hashlib.sha256()
+    with open(script_path, "rb") as f:
+        for byte_block in iter(lambda: f.read(4096), b""):
+            sha256_hash.update(byte_block)
+    return sha256_hash.hexdigest()
+
+def get_latest_migration(db: Session) -> Optional[Migration]:
+    """
+    Get the most recently applied migration.
+    
+    Args:
+        db: Database session
+        
+    Returns:
+        Migration: Latest applied migration or None if no migrations exist
+    """
+    return db.query(Migration).filter(Migration.applied == True)\
+             .order_by(desc(Migration.version)).first()
+
+def validate_migration_version(version: str) -> bool:
+    """
+    Validate migration version format (e.g. v1.0.0).
+    
+    Args:
+        version: Version string to validate
+        
+    Returns:
+        bool: True if version format is valid
+    """
+    import re
+    pattern = r'^v\d+\.\d+\.\d+$'
+    return bool(re.match(pattern, version))
+
+def mark_migration_complete(db: Session, migration: Migration, execution_time_ms: int) -> Migration:
+    """
+    Mark a migration as successfully applied.
+    
+    Args:
+        db: Database session
+        migration: Migration object to update
+        execution_time_ms: Migration execution time in milliseconds
+        
+    Returns:
+        Migration: Updated migration object
+    """
+    migration.applied = True
+    migration.execution_time = datetime.utcnow()
+    db.commit()
+    db.refresh(migration)
+    return migration
+
+def verify_migration_integrity(db: Session, script_path: str) -> Dict[str, bool]:
+    """
+    Verify migration script integrity by comparing checksums.
+    
+    Args:
+        db: Database session
+        script_path: Path to migration script
+        
+    Returns:
+        dict: Status of integrity check
+    """
+    current_checksum = calculate_script_checksum(script_path)
+    migration = db.query(Migration).filter(Migration.script_path == script_path).first()
+    
+    if not migration:
+        return {"valid": False, "error": "Migration not found in database"}
+        
+    if not migration.checksum:
+        return {"valid": False, "error": "No checksum stored for migration"}
+        
+    return {
+        "valid": current_checksum == migration.checksum,
+        "stored_checksum": migration.checksum,
+        "current_checksum": current_checksum
+    }