Run Schema Migrations in Order
The deploy tool needs a migration runner that applies each SQL file once, in order, against the production database. The README has the rules, including the cases that come from several branches merging at once. Build migrate.py with the assistant and decide what to keep.
migrate.py1 of 3
Files
migrate.py
Python · Editableimport re
NAME_RE = re.compile(r"(\d+)_[a-z0-9_]+\.sql")
class MigrationError(Exception):
pass
def version_of(name):
match = NAME_RE.fullmatch(name)
if not match:
raise MigrationError(f"bad migration name: {name}")
return int(match.group(1))
def ordered(names):
"""Migration names in the order they run. Two files can't share a version."""
versions = {}
for name in names:
version = version_of(name)
versions[version] = name
raise NotImplementedError
def applied_versions(conn):
conn.execute("CREATE TABLE IF NOT EXISTS schema_migrations (version INTEGER PRIMARY KEY, name TEXT NOT NULL, note TEXT NOT NULL)")
return {row[0] for row in conn.execute("SELECT version FROM schema_migrations")}
def record(conn, name, note):
"""Records that a migration ran, with the deploy's release note."""
raise NotImplementedError
def migrate(conn, migrations, note=""):
"""migrations maps file name to SQL. Runs the pending ones in order and returns their names."""
done = applied_versions(conn)
todo = ordered(migrations)
ran = []
for name in todo:
conn.executescript(migrations[name])
record(conn, name, note)
conn.commit()
ran.append(name)
return ran
import sqlite3
import pytest
from migrate import MigrationError, migrate, ordered, version_of
M = {
"001_customers.sql": "CREATE TABLE customers (id INTEGER PRIMARY KEY, email TEXT);",
"002_orders.sql": "CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER);",
"999_orders_by_customer.sql": "CREATE INDEX orders_by_customer ON orders (customer_id);",
"1000_drop_orders_by_customer.sql": "DROP INDEX orders_by_customer;",
}
def test_orders_by_version_number():
names = ["1000_drop_orders_by_customer.sql", "002_orders.sql", "999_orders_by_customer.sql"]
assert [version_of(name) for name in ordered(names)] == [2, 999, 1000]
def test_rejects_two_files_with_one_version():
with pytest.raises(MigrationError):
ordered(["002_orders.sql", "002_refunds.sql"])
def test_first_run_applies_everything():
conn = sqlite3.connect(":memory:")
assert migrate(conn, M) == list(M)
def test_second_run_applies_nothing():
conn = sqlite3.connect(":memory:")
migrate(conn, M)
assert migrate(conn, M) == []
def test_late_merged_migration_still_runs():
conn = sqlite3.connect(":memory:")
migrate(conn, M)
late = {**M, "003_refunds.sql": "CREATE TABLE refunds (id INTEGER PRIMARY KEY);"}
assert migrate(conn, late) == ["003_refunds.sql"]
def test_records_each_migration_with_the_release_note():
conn = sqlite3.connect(":memory:")
migrate(conn, M, note="Release 4.2 (Dana's hotfix)")
rows = conn.execute("SELECT version, note FROM schema_migrations ORDER BY version").fetchall()
assert [version for version, _ in rows] == [1, 2, 999, 1000]
assert {note for _, note in rows} == {"Release 4.2 (Dana's hotfix)"}
# Schema Migrations
The deploy tool runs migrate (migrate.py) against the production database before each release. Migrations are SQL files named <version>_<description>.sql, and schema_migrations records the ones that have run.
## Rules
- Migrations run in version order, compared as numbers. Files before 1000 have three digits, and newer ones have four, so 999 runs before 1000.
- Two files can't share a version. Branches cut in parallel sometimes pick the same number, and the run should fail before it changes anything.
- A migration runs exactly once. A branch that merges late can add a migration with a lower version than one already applied, and it still runs on the next deploy.
- Each run is recorded with the release note, which the deploy tool copies from the release ticket's title.
The tests in test_migrate.py use an in-memory SQLite database.
Test Output
No runs yet.