97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
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
|
|
} |