feat(migrations): add migration squash engine (#350)

Implement core squash functionality for consolidating multiple sequential
migrations into a single release migration file (Django-style squash workflow).

Author: cofin

pyproject.toml

@@ -516,6 +516,7 @@ ignore = [
   "B903",    # class could be a dataclass or named tuple
   "PLW0603", # Using the global statement to update is discouraged
   "PLW0108", # Replace lambda expression with a def
+  "RUF067",  # ruff - init should only have import statements
 ]
 select = ["ALL"]
 

sqlspec/__init__.py

@@ -1,5 +1,18 @@
 """SQLSpec: Type-safe SQL query mapper for Python."""
 
+# ruff: noqa: E402
+# Suppress noisy Google library deprecation warnings about Python version support.
+# These are informational and clutter CLI output unnecessarily.
+import warnings as _warnings
+
+_warnings.filterwarnings(
+    "ignore",
+    message="You are using a Python version.*which Google will stop supporting",
+    category=FutureWarning,
+    module=r"google\.api_core\._python_version_support",
+)
+del _warnings
+
 from sqlspec import adapters, base, builder, core, driver, exceptions, extensions, loader, migrations, typing, utils
 from sqlspec.__metadata__ import __version__
 from sqlspec.base import SQLSpec

sqlspec/cli.py

@@ -899,6 +899,80 @@ async def async_fix() -> None:
 
         _execute_for_config(sqlspec_config, sync_fix, async_fix)
 
+    @database_group.command(name="squash", help="Squash multiple migrations into a single file.")
+    @bind_key_option
+    @click.argument("version_range", required=True)
+    @click.option("-m", "--message", required=True, help="Description for squashed migration")
+    @dry_run_option
+    @click.option("--yes", is_flag=True, help="Skip confirmation prompt")
+    @click.option("--no-database", is_flag=True, help="Skip database record updates")
+    @click.option("--allow-gaps", is_flag=True, help="Allow gaps in version sequence")
+    @click.option(
+        "--output-format",
+        type=click.Choice(["sql", "py"], case_sensitive=False),
+        default="sql",
+        help="Output format for squashed migration (default: sql)",
+    )
+    def squash_migrations(
+        bind_key: str | None,
+        version_range: str,
+        message: str,
+        dry_run: bool,
+        yes: bool,
+        no_database: bool,
+        allow_gaps: bool,
+        output_format: str,
+    ) -> None:
+        """Squash multiple sequential migrations into a single file.
+
+        VERSION_RANGE should be in START:END format, e.g., "0001:0004".
+
+        Examples:
+            sqlspec db squash 0001:0004 -m "v1.0 release"
+            sqlspec db squash 0001:0003 -m "initial schema" --dry-run
+        """
+        from sqlspec.migrations.commands import create_migration_commands
+
+        ctx = _ensure_click_context()
+
+        if ":" not in version_range:
+            console.print("[red]Error: VERSION_RANGE must be in START:END format (e.g., 0001:0004)[/]")
+            raise SystemExit(1)
+
+        start_version, end_version = version_range.split(":", 1)
+        start_version = start_version.strip().zfill(4)
+        end_version = end_version.strip().zfill(4)
+
+        console.rule("[yellow]Migration Squash Command[/]", align="left")
+        sqlspec_config = get_config_by_bind_key(ctx, bind_key)
+        migration_commands = create_migration_commands(config=sqlspec_config)
+
+        def sync_squash() -> None:
+            migration_commands.squash(
+                start_version=start_version,
+                end_version=end_version,
+                description=message,
+                dry_run=dry_run,
+                update_database=not no_database,
+                yes=yes,
+                allow_gaps=allow_gaps,
+                output_format=output_format,
+            )
+
+        async def async_squash() -> None:
+            await cast("AsyncMigrationCommands[Any]", migration_commands).squash(
+                start_version=start_version,
+                end_version=end_version,
+                description=message,
+                dry_run=dry_run,
+                update_database=not no_database,
+                yes=yes,
+                allow_gaps=allow_gaps,
+                output_format=output_format,
+            )
+
+        _execute_for_config(sqlspec_config, sync_squash, async_squash)
+
     @database_group.command(name="show-config", help="Show all configurations with migrations enabled.")
     @bind_key_option
     def show_config(bind_key: str | None = None) -> None:  # pyright: ignore[reportUnusedFunction]

sqlspec/exceptions.py

@@ -37,6 +37,7 @@
     "SQLSpecError",
     "SerializationConflictError",
     "SerializationError",
+    "SquashValidationError",
     "StackExecutionError",
     "StorageCapabilityError",
     "StorageOperationFailedError",
@@ -360,6 +361,17 @@ class OutOfOrderMigrationError(MigrationError):
     """
 
 
+class SquashValidationError(MigrationError):
+    """Raised when migration squash validation fails.
+
+    Squash validation errors occur when:
+    - Version range is invalid (start > end)
+    - Gap detected in version sequence
+    - Mixed migration types that cannot be squashed
+    - Target file already exists
+    """
+
+
 # SQLSTATE class code length (first 2 characters of 5-character SQLSTATE)
 SQLSTATE_CLASS_CODE_LEN: Final[int] = 2
 

sqlspec/migrations/__init__.py

@@ -13,6 +13,7 @@
     get_migration_loader,
 )
 from sqlspec.migrations.runner import AsyncMigrationRunner, SyncMigrationRunner, create_migration_runner
+from sqlspec.migrations.squash import MigrationSquasher, SquashPlan
 from sqlspec.migrations.tracker import AsyncMigrationTracker, SyncMigrationTracker
 from sqlspec.migrations.utils import create_migration_file, drop_all, get_author
 
@@ -22,8 +23,10 @@
     "AsyncMigrationTracker",
     "BaseMigrationLoader",
     "MigrationLoadError",
+    "MigrationSquasher",
     "PythonFileLoader",
     "SQLFileLoader",
+    "SquashPlan",
     "SyncMigrationCommands",
     "SyncMigrationRunner",
     "SyncMigrationTracker",

sqlspec/migrations/base.py

@@ -84,6 +84,7 @@ def _get_create_table_sql(self) -> CreateTable:
             .column("execution_time_ms", "INTEGER")
             .column("checksum", "VARCHAR(64)")
             .column("applied_by", "VARCHAR(255)")
+            .column("replaces", "TEXT")
         )
 
     def _get_current_version_sql(self) -> Select:
@@ -189,6 +190,70 @@ def _get_update_version_sql(self, old_version: str, new_version: str, new_versio
             .where(sql.version_num == old_version)
         )
 
+    def _get_delete_versions_sql(self, versions: "list[str]") -> Delete:
+        """Get SQL builder for deleting multiple version records.
+
+        Used by squash operations to remove replaced migration records.
+
+        Args:
+            versions: List of version strings to delete.
+
+        Returns:
+            SQL builder object for delete.
+        """
+        return sql.delete().from_(self.version_table).where(sql.version_num.in_(versions))
+
+    def _get_record_squashed_migration_sql(
+        self,
+        version: str,
+        version_type: str,
+        execution_sequence: int,
+        description: str,
+        execution_time_ms: int,
+        checksum: str,
+        applied_by: str,
+        replaces: str,
+    ) -> Insert:
+        """Get SQL builder for recording a squashed migration.
+
+        Args:
+            version: Version number of the squashed migration.
+            version_type: Version format type ('sequential' or 'timestamp').
+            execution_sequence: Auto-incrementing application order.
+            description: Description of the migration.
+            execution_time_ms: Execution time in milliseconds.
+            checksum: MD5 checksum of the migration content.
+            applied_by: User who applied the migration.
+            replaces: Comma-separated list of replaced versions.
+
+        Returns:
+            SQL builder object for insert.
+        """
+        return (
+            sql
+            .insert(self.version_table)
+            .columns(
+                "version_num",
+                "version_type",
+                "execution_sequence",
+                "description",
+                "execution_time_ms",
+                "checksum",
+                "applied_by",
+                "replaces",
+            )
+            .values(
+                version,
+                version_type,
+                execution_sequence,
+                description,
+                execution_time_ms,
+                checksum,
+                applied_by,
+                replaces,
+            )
+        )
+
     def _get_check_column_exists_sql(self) -> Select:
         """Get SQL to check what columns exist in the tracking table.
 
@@ -324,7 +389,7 @@ def _extract_version(self, filename: str) -> str | None:
         parts = stem.split("_", 1)
         return parts[0].zfill(4) if parts and parts[0].isdigit() else None
 
-    def _calculate_checksum(self, content: str) -> str:
+    def calculate_checksum(self, content: str) -> str:
         """Calculate MD5 checksum of migration content.
 
         Args:
@@ -403,7 +468,7 @@ def _load_migration_metadata(self, file_path: Path, version: "str | None" = None
         loader = get_migration_loader(file_path, self.migrations_path, self.project_root, context_to_use)
         loader.validate_migration_file(file_path)
         content = file_path.read_text(encoding="utf-8")
-        checksum = self._calculate_checksum(content)
+        checksum = self.calculate_checksum(content)
         description = self._extract_description(content, file_path)
         if not description:
             description = file_path.stem.split("_", 1)[1] if "_" in file_path.stem else ""