SQLite 规则快照示例：改配送费后，旧订单为什么不该跟着变

import sqlite3
from contextlib import closing


def publish(db, version, fee, can_cancel):
    with db:
        db.execute("INSERT INTO rule VALUES (?, ?, ?)",
                   (version, fee, int(can_cancel)))
        db.execute("UPDATE active_rule SET version=? WHERE id=1", (version,))


def create_order(db, order_id):
    with db:
        db.execute("BEGIN IMMEDIATE")
        version, fee = db.execute(
            "SELECT r.version, r.fee_fen FROM active_rule a "
            "JOIN rule r ON r.version=a.version WHERE a.id=1"
        ).fetchone()
        db.execute("INSERT INTO orders VALUES (?, ?, ?, 'new', 1)",
                   (order_id, version, fee))


def transition(db, order_id, expected, action):
    if action not in {"accept", "cancel"}:
        raise ValueError("unknown action")
    with db:
        row = db.execute(
            "SELECT r.can_cancel FROM orders o "
            "JOIN rule r ON r.version=o.rule_version WHERE o.id=?",
            (order_id,)
        ).fetchone()
        if row is None:
            raise ValueError("unknown order")
        if action == "cancel" and not row[0]:
            return "policy_denied"
        state = "accepted" if action == "accept" else "cancelled"
        changed = db.execute(
            "UPDATE orders SET state=?, version=version+1 "
            "WHERE id=? AND state='new' AND version=?",
            (state, order_id, expected)
        ).rowcount
        return "applied" if changed else "conflict"


def main():
    with closing(sqlite3.connect(":memory:")) as db:
        db.executescript("""
        PRAGMA foreign_keys=ON;
        CREATE TABLE rule (version INTEGER PRIMARY KEY, fee_fen INTEGER CHECK(fee_fen>=0),
          can_cancel INTEGER CHECK(can_cancel IN (0,1)));
        CREATE TRIGGER immutable_rule_update BEFORE UPDATE ON rule
          BEGIN SELECT RAISE(ABORT, 'published rule is immutable'); END;
        CREATE TRIGGER immutable_rule_delete BEFORE DELETE ON rule
          BEGIN SELECT RAISE(ABORT, 'published rule is immutable'); END;
        INSERT INTO rule VALUES (1, 300, 1);
        CREATE TABLE active_rule (id INTEGER PRIMARY KEY, version INTEGER REFERENCES rule);
        INSERT INTO active_rule VALUES (1, 1);
        CREATE TABLE orders (id TEXT PRIMARY KEY, rule_version INTEGER REFERENCES rule,
          fee_fen INTEGER, state TEXT, version INTEGER);
        """)
        create_order(db, "old")
        publish(db, 2, 500, False)
        create_order(db, "new")
        rows = db.execute(
            "SELECT id, rule_version, fee_fen FROM orders ORDER BY rowid"
        ).fetchall()
        assert rows == [("old", 1, 300), ("new", 2, 500)]
        assert transition(db, "old", 1, "cancel") == "applied"
        assert transition(db, "new", 1, "cancel") == "policy_denied"
        assert transition(db, "new", 1, "accept") == "applied"
        assert transition(db, "new", 1, "accept") == "conflict"
        try:
            with db:
                db.execute("UPDATE rule SET fee_fen=999 WHERE version=1")
        except sqlite3.IntegrityError:
            pass
        else:
            raise AssertionError("published rule was overwritten")
        assert db.execute("SELECT fee_fen FROM rule WHERE version=1").fetchone()[0] == 300
        print("snapshots:", rows)
        print("old order cancellation: applied")
        print("new order cancellation: policy_denied")
        print("stale order version: conflict")
        print("published rule mutation: rejected")


if __name__ == "__main__":
    main()

python demo.py

snapshots: [('old', 1, 300), ('new', 2, 500)]
old order cancellation: applied
new order cancellation: policy_denied
stale order version: conflict
published rule mutation: rejected