From 0bad21280fc29e42a58358a68f702cd270bb9094 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 19:08:53 +0000 Subject: [PATCH] migrate --stop-before: validate names and fix legacy compatibility An unknown --stop-before value previously matched nothing and every migration was silently applied, including the one the user meant to stop before. The CLI now errors unless each value matches a known migration. The CLI also always passed stop_before as a list, but older duck-typed sqlite-migrate Migrations objects compare it against a single string name - so the flag was silently ignored for exactly the migration files the compatibility shim exists to support. Legacy sets now receive a single string, with an error if multiple values target one legacy set. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd --- docs/migrations.rst | 2 + sqlite_utils/cli.py | 35 +++++++++-- tests/test_cli_migrate.py | 125 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 158 insertions(+), 4 deletions(-) diff --git a/docs/migrations.rst b/docs/migrations.rst index 5628ded..eddac5f 100644 --- a/docs/migrations.rst +++ b/docs/migrations.rst @@ -157,6 +157,8 @@ You can also target a specific migration set using ``migration_set:migration_nam The ``--stop-before`` option can be passed more than once. +If a ``--stop-before`` value does not match any known migration the command exits with an error, rather than silently applying everything. + Verbose output ============== diff --git a/sqlite_utils/cli.py b/sqlite_utils/cli.py index d51ba7b..23f04a7 100644 --- a/sqlite_utils/cli.py +++ b/sqlite_utils/cli.py @@ -3400,11 +3400,38 @@ def migrate(db_path, migrations, stop_before, list_, verbose): click.echo("\nSchema before:\n") click.echo(textwrap.indent(prev_schema, " ") or " (empty)") click.echo() + if stop_before: + # Every --stop-before value must match at least one known migration + known_names = set() + for migration_set in migration_sets: + names = {m.name for m in migration_set.pending(db)} + names.update(m.name for m in migration_set.applied(db)) + known_names.update(names) + known_names.update( + "{}:{}".format(migration_set.name, name) for name in names + ) + unknown = [value for value in stop_before if value not in known_names] + if unknown: + raise click.ClickException( + "--stop-before did not match any migrations: {}".format( + ", ".join(unknown) + ) + ) for migration_set in migration_sets: - migration_set.apply( - db, - stop_before=_stop_before_for_migration_set(stop_before, migration_set.name), - ) + matches = _stop_before_for_migration_set(stop_before, migration_set.name) + if isinstance(migration_set, sqlite_utils.Migrations): + migration_set.apply(db, stop_before=matches) + else: + # Legacy sqlite-migrate Migrations objects take a single string + # for stop_before, not a list + distinct = list(dict.fromkeys(matches)) + if len(distinct) > 1: + raise click.ClickException( + "Migration set '{}' uses the older sqlite-migrate class, " + "which only supports a single --stop-before value - " + "got: {}".format(migration_set.name, ", ".join(distinct)) + ) + migration_set.apply(db, stop_before=distinct[0] if distinct else None) if verbose: click.echo("Schema after:\n") post_schema = db.schema diff --git a/tests/test_cli_migrate.py b/tests/test_cli_migrate.py index 229bc9c..4a2ca9c 100644 --- a/tests/test_cli_migrate.py +++ b/tests/test_cli_migrate.py @@ -301,3 +301,128 @@ def test_stop_before_multiple_qualified(two_sets_same_migration_name): assert not db["creature_weights"].exists() assert db["sales"].exists() assert not db["sales_weights"].exists() + + +LEGACY_MIGRATIONS = """ +import datetime + +class _Migration: + def __init__(self, name, fn): + self.name = name + self.fn = fn + +class _Applied: + def __init__(self, name, applied_at): + self.name = name + self.applied_at = applied_at + +class LegacyMigrations: + # Mimics the sqlite-migrate 0.x Migrations class, in particular + # apply(db, stop_before=None) taking a single string + migrations_table = "_sqlite_migrations" + + def __init__(self, name): + self.name = name + self._migrations = [] + + def __call__(self, fn): + self._migrations.append(_Migration(fn.__name__, fn)) + return fn + + def ensure_migrations_table(self, db): + db[self.migrations_table].create( + {"migration_set": str, "name": str, "applied_at": str}, + pk=("migration_set", "name"), + if_not_exists=True, + ) + + def applied(self, db): + self.ensure_migrations_table(db) + return [ + _Applied(row["name"], row["applied_at"]) + for row in db[self.migrations_table].rows_where( + "migration_set = ?", [self.name] + ) + ] + + def pending(self, db): + applied = {m.name for m in self.applied(db)} + return [m for m in self._migrations if m.name not in applied] + + def apply(self, db, stop_before=None): + for migration in self.pending(db): + if migration.name == stop_before: + return + migration.fn(db) + db[self.migrations_table].insert( + { + "migration_set": self.name, + "name": migration.name, + "applied_at": str( + datetime.datetime.now(datetime.timezone.utc) + ), + } + ) + +legacy = LegacyMigrations("legacy_set") + +@legacy +def first(db): + db["first"].insert({"hello": "world"}) + +@legacy +def second(db): + db["second"].insert({"hello": "world"}) +""" + + +def test_stop_before_unknown_name_errors(two_migrations): + path, _ = two_migrations + db_path = str(path / "test.db") + runner = CliRunner() + result = runner.invoke( + sqlite_utils.cli.cli, + ["migrate", db_path, str(path), "--stop-before", "fooo"], + ) + assert result.exit_code == 1 + assert "--stop-before did not match any migrations: fooo" in result.output + # Nothing should have been applied + db = sqlite_utils.Database(db_path) + assert "foo" not in db.table_names() + assert "bar" not in db.table_names() + + +def test_stop_before_with_legacy_migrations_class(tmpdir): + path = pathlib.Path(tmpdir) + (path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8") + db_path = str(path / "test.db") + runner = CliRunner() + result = runner.invoke( + sqlite_utils.cli.cli, + ["migrate", db_path, str(path), "--stop-before", "second"], + ) + assert result.exit_code == 0, result.output + db = sqlite_utils.Database(db_path) + assert "first" in db.table_names() + assert "second" not in db.table_names() + + +def test_stop_before_multiple_values_for_legacy_set_errors(tmpdir): + path = pathlib.Path(tmpdir) + (path / "migrations.py").write_text(LEGACY_MIGRATIONS, "utf-8") + db_path = str(path / "test.db") + runner = CliRunner() + result = runner.invoke( + sqlite_utils.cli.cli, + [ + "migrate", + db_path, + str(path), + "--stop-before", + "legacy_set:first", + "--stop-before", + "legacy_set:second", + ], + ) + assert result.exit_code == 1 + assert "single --stop-before" in result.output