From 8665adb695c5dc18fb995553192b6ea7486aeb98 Mon Sep 17 00:00:00 2001
From: Samran Asif <samranwebdev2000@gmail.com>
Date: Mon, 21 Sep 2026 19:59:52 +0500
Subject: [PATCH] migrate: pre-validate --stop-before targets across all sets
 before applying

When sqlite-utils migrate is invoked with multiple migration sets, an already-applied --stop-before target in a subsequent set would cause an error inside that set's apply() call only after earlier sets had already executed and committed their pending migrations.

Pre-validate across all loaded migration sets upfront that no requested --stop-before targets have already been applied before entering the application loop, ensuring no pending migrations are applied when an error is raised.

Fixes #870

Signed-off-by: Samran Asif <samranwebdev2000@gmail.com>
---
 sqlite_utils/cli.py       | 14 ++++++++++++++
 tests/test_cli_migrate.py | 35 +++++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+)

diff --git a/sqlite_utils/cli.py b/sqlite_utils/cli.py
index c23090283..49a798290 100644
--- a/sqlite_utils/cli.py
+++ b/sqlite_utils/cli.py
@@ -3598,6 +3598,20 @@ def migrate(db_path, migrations, stop_before, list_, verbose):
                     ", ".join(unknown)
                 )
             )
+        for migration_set in migration_sets:
+            matches = _stop_before_for_migration_set(stop_before, migration_set.name)
+            if hasattr(migration_set, "applied"):
+                applied_names = {m.name for m in migration_set.applied(db)}
+                already_applied = set(matches).intersection(applied_names)
+                if already_applied:
+                    raise click.ClickException(
+                        "Cannot stop before migration{} {} in set '{}' - already "
+                        "been applied".format(
+                            "s" if len(already_applied) > 1 else "",
+                            ", ".join(sorted(already_applied)),
+                            migration_set.name,
+                        )
+                    )
     for migration_set in migration_sets:
         matches = _stop_before_for_migration_set(stop_before, migration_set.name)
         if isinstance(migration_set, sqlite_utils.Migrations):
diff --git a/tests/test_cli_migrate.py b/tests/test_cli_migrate.py
index 743988736..8f08764ea 100644
--- a/tests/test_cli_migrate.py
+++ b/tests/test_cli_migrate.py
@@ -506,3 +506,38 @@ def test_list_with_legacy_class_is_read_only(tmpdir):
     db2 = sqlite_utils.Database(db_path)
     assert "_sqlite_migrations" not in db2.table_names()
     db2.close()
+
+
+def test_stop_before_applied_migration_multiple_sets_does_not_apply_earlier(tmpdir):
+    root = pathlib.Path(tmpdir)
+    for name in ("a", "b"):
+        (root / f"{name}.py").write_text(
+            "from sqlite_utils import Migrations\n"
+            f'migrations = Migrations("{name}")\n'
+            "@migrations()\n"
+            "def first(db):\n"
+            f'    db.table("{name}").insert({{"value": 1}})\n',
+            "utf-8",
+        )
+    db_path = str(root / "test.db")
+    runner = CliRunner()
+    # Apply set b first
+    seed = runner.invoke(sqlite_utils.cli.cli, ["migrate", db_path, str(root / "b.py")])
+    assert seed.exit_code == 0
+    # Now migrate both a and b with --stop-before b:first (which is already applied).
+    # This must error and must NOT apply set a.
+    result = runner.invoke(
+        sqlite_utils.cli.cli,
+        [
+            "migrate",
+            db_path,
+            str(root / "a.py"),
+            str(root / "b.py"),
+            "--stop-before",
+            "b:first",
+        ],
+    )
+    assert result.exit_code != 0
+    assert "already been applied" in result.output
+    db = sqlite_utils.Database(db_path)
+    assert not db.table("a").exists()