迁移是可版本化的 schema 变更

"""Simulate a versioned migration chain with an applied ledger (stdlib only)."""

# Each migration is (version, name, up, down). down=None means irreversible.
# The `applied` dict emulates a schema_migrations table: version -> a name-based identity (not a content digest).

def identity(version, name):
    return "%d:%s" % (version, name)


def make_migrations():
    db = {"users": []}

    def up1():
        db["users"] = []

    def down1():
        del db["users"]

    def up2():
        for user in db["users"]:
            user["email"] = None
        db["has_email_column"] = True

    def down2():
        # Irreversible in effect: dropping the column destroys its data.
        for user in db["users"]:
            del user["email"]
        del db["has_email_column"]

    def up3():
        db["index_on_email"] = True

    def down3():
        del db["index_on_email"]

    migrations = [
        (1, "create_users", up1, down1),
        (2, "add_email_column", up2, down2),
        (3, "add_email_index", up3, down3),
    ]
    return db, migrations


def apply_single(db, applied, migrations, version):
    by_version = {m[0]: m for m in migrations}
    assert version in by_version, "unknown migration version: %r" % (version,)
    wanted = by_version[version]
    if version in applied:
        if applied[version] != identity(version, wanted[1]):
            raise RuntimeError("version drift: v%d was edited after apply" % version)
        return "skipped"  # idempotent re-run
    missing = [m[0] for m in migrations if m[0] < version and m[0] not in applied]
    if missing:
        raise RuntimeError("out of order: v%d needs missing %s first" % (version, missing))
    wanted[2]()
    applied[version] = identity(version, wanted[1])
    return "applied"


def migrate_up(db, applied, migrations):
    results = []
    for m in migrations:
        results.append((m[0], apply_single(db, applied, migrations, m[0])))
    return results


def migrate_down(db, applied, migrations, target):
    by_version = {m[0]: m for m in migrations}
    while applied and max(applied) > target:
        version = max(applied)
        migration = by_version[version]
        assert migration[3] is not None, "v%d is irreversible" % version
        migration[3]()
        del applied[version]
    return sorted(applied)


# 1. Fresh database replays the whole chain in order.
db, migrations = make_migrations()
applied = {}
assert migrate_up(db, applied, migrations) == [(1, "applied"), (2, "applied"), (3, "applied")]
assert sorted(applied) == [1, 2, 3] and db["has_email_column"] is True
print("fresh replay applies v1..v3 in order:", sorted(applied))

# 2. Re-running is idempotent: the applied ledger makes it a no-op.
assert migrate_up(db, applied, migrations) == [(1, "skipped"), (2, "skipped"), (3, "skipped")]
assert sorted(applied) == [1, 2, 3]
print("second run is a no-op; applied ledger still:", sorted(applied))

# 3. Out-of-order apply is rejected instead of silently forking history.
db2, migrations2 = make_migrations()
applied2 = {}
try:
    apply_single(db2, applied2, migrations2, 3)
except RuntimeError as error:
    assert "out of order" in str(error), error
    print("out-of-order v3 rejected:", error)
else:
    raise AssertionError("applying v3 before v1..v2 must fail")
assert applied2 == {} and "index_on_email" not in db2

# 4. Rollback runs downs in reverse and only to the requested target.
assert migrate_down(db, applied, migrations, 2) == [1, 2]
assert "index_on_email" not in db and sorted(applied) == [1, 2]
assert migrate_up(db, applied, migrations) == [(1, "skipped"), (2, "skipped"), (3, "applied")]
print("rollback to v2 drops the index; re-up restores it:", sorted(applied))

# 5. A down migration destroys data: emails do not come back on re-up.
db["users"].append({"id": 1, "email": "a@example.com"})
assert migrate_down(db, applied, migrations, 1) == [1]
assert db["users"] == [{"id": 1}]  # email column (and its values) is gone
assert migrate_up(db, applied, migrations)[1] == (2, "applied")
assert db["users"] == [{"id": 1, "email": None}]  # re-up gives NULL, not the old address
print("down v2 drops stored emails; re-up yields None instead of the original")

# 6. Editing an already-applied migration is version drift, not a fix.
drifted = [(v, n + "_renamed" if v == 2 else n, u, d) for v, n, u, d in migrations]
try:
    migrate_up(db, applied, drifted)
except RuntimeError as error:
    assert "drift" in str(error), error
    print("edited history detected:", error)
else:
    raise AssertionError("renaming applied v2 must fail")

print("PASS: ordered up, idempotent ledger, rejected jump, ordered rollback, lossy down, drift check")