外卖跑腿系统选型：报价变更的版本条件更新与确认约束

import sqlite3,tempfile,json
from pathlib import Path
with tempfile.TemporaryDirectory() as folder:
    path=Path(folder)/'quote.db'
    a=sqlite3.connect(path); b=sqlite3.connect(path)
    a.executescript('''
    CREATE TABLE quote(id TEXT PRIMARY KEY, customer TEXT NOT NULL,
      revision INTEGER NOT NULL CHECK(typeof(revision)='integer' AND revision>0),
      goods INTEGER NOT NULL CHECK(typeof(goods)='integer' AND goods>=0),
      service INTEGER NOT NULL CHECK(typeof(service)='integer' AND service>=0),
      state TEXT NOT NULL CHECK(state IN ('offered','confirmed','cancelled')));
    CREATE TABLE confirmation(order_id TEXT NOT NULL, revision INTEGER NOT NULL,
      customer TEXT NOT NULL, goods INTEGER NOT NULL, service INTEGER NOT NULL,
      PRIMARY KEY(order_id,revision));
    INSERT INTO quote VALUES('B1','U1',1,3200,800,'offered');
    INSERT INTO quote VALUES('B2','U1',1,0,600,'cancelled');
    ''')
    def confirm(db,principal,order_id,revision):
        # principal is supplied by authenticated server middleware.
        if type(revision) is not int or revision<=0:return {'status':422}
        with db:
            row=db.execute('SELECT customer,revision,goods,service,state FROM quote WHERE id=?',
                           (order_id,)).fetchone()
            if row is None:return {'status':404}
            customer,current,goods,service,state=row
            if customer!=principal:return {'status':403}
            if current!=revision or state=='cancelled':return {'status':409}
            if state=='confirmed':return {'status':200,'duplicate':True}
            changed=db.execute('''UPDATE quote SET state='confirmed'
              WHERE id=? AND customer=? AND revision=? AND state='offered' ''',
              (order_id,principal,revision)).rowcount
            if changed!=1:return {'status':409}
            db.execute('INSERT INTO confirmation VALUES(?,?,?,?,?)',
                       (order_id,revision,principal,goods,service))
            return {'status':200,'duplicate':False}
    checks=0
    def check(actual,expected):
        global checks
        assert actual==expected,(actual,expected)
        checks+=1
    check(confirm(a,'U2','B1',1),{'status':403})
    check(confirm(a,'U1','missing',1),{'status':404})
    check(confirm(a,'U1','B1',True),{'status':422})
    check(confirm(a,'U1','B1',0),{'status':422})
    # Both clients read v1; a controlled proposal then publishes v2.
    old_version=b.execute("SELECT revision FROM quote WHERE id='B1'").fetchone()[0]
    with a:
        a.execute("UPDATE quote SET revision=2,service=1000 WHERE id='B1' AND revision=1")
    check(confirm(b,'U1','B1',old_version),{'status':409})
    check(a.execute('SELECT count(*) FROM confirmation').fetchone()[0],0)
    check(confirm(a,'U1','B1',2),{'status':200,'duplicate':False})
    check(confirm(b,'U1','B1',2),{'status':200,'duplicate':True})
    check(a.execute('SELECT count(*) FROM confirmation').fetchone()[0],1)
    check(a.execute('SELECT goods,service FROM confirmation').fetchone(),(3200,1000))
    check(confirm(a,'U1','B2',1),{'status':409})
    check(confirm(b,'U2','B1',2),{'status':403})
    try:
        with a:a.execute("UPDATE quote SET service=-1 WHERE id='B1'")
    except sqlite3.IntegrityError:constraint=True
    else:constraint=False
    check(constraint,True)
    check(a.execute("SELECT service FROM quote WHERE id='B1'").fetchone()[0],1000)
    a.close();b.close()
    print(json.dumps({'checks':checks,'staleWriter':409,'confirmations':1,
      'scope':'local sequential two-connection example; no production concurrency or payment'}))