免费外卖系统选型中的商家额度校验：事务与重复请求示例

import sqlite3

def connect():
    db = sqlite3.connect(':memory:', isolation_level=None)
    db.executescript('''
    CREATE TABLE quota(tenant TEXT PRIMARY KEY, maximum INTEGER NOT NULL CHECK(maximum>=0));
    CREATE TABLE merchant(tenant TEXT NOT NULL, request_id TEXT NOT NULL,
      name TEXT NOT NULL, PRIMARY KEY(tenant,request_id));
    INSERT INTO quota VALUES('demo',2);
    ''')
    return db

def create(db, tenant, request_id, name):
    if not all(isinstance(x,str) and x.strip() for x in (tenant,request_id,name)):
        raise ValueError('invalid input')
    db.execute('BEGIN IMMEDIATE')
    try:
        prior = db.execute('SELECT name FROM merchant WHERE tenant=? AND request_id=?',
                           (tenant,request_id)).fetchone()
        if prior:
            if prior[0] != name:
                raise ValueError('request conflict')
            db.commit()
            return 'existing'
        quota = db.execute('SELECT maximum FROM quota WHERE tenant=?',(tenant,)).fetchone()
        if quota is None:
            raise ValueError('unknown tenant')
        count = db.execute('SELECT count(*) FROM merchant WHERE tenant=?',(tenant,)).fetchone()[0]
        if count >= quota[0]:
            db.commit()
            return 'quota_reached'
        db.execute('INSERT INTO merchant VALUES(?,?,?)',(tenant,request_id,name))
        db.commit()
        return 'created'
    except Exception:
        db.rollback()
        raise

def expect_error(fn, message):
    try:
        fn()
    except ValueError as error:
        assert str(error) == message
    else:
        raise AssertionError('expected failure')

db=connect()
assert create(db,'demo','r1','shop A') == 'created'
assert create(db,'demo','r1','shop A') == 'existing'
expect_error(lambda:create(db,'demo','r1','other'),'request conflict')
assert create(db,'demo','r2','shop B') == 'created'
assert create(db,'demo','r3','shop C') == 'quota_reached'
assert create(db,'demo','r2','shop B') == 'existing'
expect_error(lambda:create(db,'missing','r1','A'),'unknown tenant')
expect_error(lambda:create(db,'demo','','A'),'invalid input')
assert db.execute('SELECT count(*) FROM merchant').fetchone()[0] == 2
print('9 local assertions passed; quota=2; created=2; retry=existing; third=quota_reached')