mirror of
https://github.com/simonw/sqlite-utils.git
synced 2026-08-02 07:24:21 +02:00
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UnLnhsH25Nnv7LHhekUfPd
This commit is contained in:
parent
fd867282b3
commit
0bad21280f
3 changed files with 158 additions and 4 deletions
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue