Source code for sqlspec.migrations.fix

"""Migration file fix operations for converting timestamp to sequential versions.

This module provides utilities to convert timestamp-format migration files to
sequential format, supporting the hybrid versioning workflow where development
uses timestamps and production uses sequential numbers.
"""

import re
from dataclasses import dataclass
from pathlib import Path

from sqlspec.migrations._backup import create_backup, remove_backup, restore_backup
from sqlspec.utils.logging import get_logger

__all__ = ("MigrationFixer", "MigrationRename")

logger = get_logger(__name__)



[docs]
@dataclass
class MigrationRename:
    """Represents a planned migration file rename operation.

    Attributes:
        old_path: Current file path.
        new_path: Target file path after rename.
        old_version: Current version string.
        new_version: Target version string.
        needs_content_update: Whether file content needs updating.
            True for SQL files that contain query names.
    """

    old_path: Path
    new_path: Path
    old_version: str
    new_version: str
    needs_content_update: bool

[docs]

class MigrationFixer: """Handles atomic migration file conversion operations. Provides backup/rollback functionality and manages conversion from timestamp-based migration files to sequential format. """

[docs]

def __init__(self, migrations_path: Path) -> None: """Initialize migration fixer. Args: migrations_path: Path to migrations directory. """ self.migrations_path = migrations_path self.backup_path: Path | None = None

[docs]

def plan_renames(self, conversion_map: dict[str, str]) -> list[MigrationRename]: """Plan all file rename operations from conversion map. Scans migration directory and builds list of MigrationRename objects for all files that need conversion. Validates no target collisions. Args: conversion_map: Dictionary mapping old versions to new versions. Returns: List of planned rename operations. Raises: ValueError: If target file already exists or collision detected. """ if not conversion_map: return [] renames: list[MigrationRename] = [] for old_version, new_version in conversion_map.items(): matching_files = list(self.migrations_path.glob(f"{old_version}_*")) for old_path in matching_files: suffix = old_path.suffix description = old_path.stem.removeprefix(f"{old_version}_") new_filename = f"{new_version}_{description}{suffix}" new_path = self.migrations_path / new_filename if new_path.exists() and new_path != old_path: msg = f"Target file already exists: {new_path}" raise ValueError(msg) needs_content_update = suffix == ".sql" renames.append( MigrationRename( old_path=old_path, new_path=new_path, old_version=old_version, new_version=new_version, needs_content_update=needs_content_update, ) ) return renames

[docs]

def create_backup(self) -> Path: """Create timestamped backup directory with all migration files. Returns: Path to created backup directory. """ self.backup_path = create_backup(self.migrations_path) return self.backup_path