Complete worked example
python
import sqlite3
def connect(path):
connection = sqlite3.connect(path, autocommit=True)
connection.execute('PRAGMA synchronous=FULL')
connection.autocommit = False
return connection
def prepare(path):
connection = connect(path)
try:
connection.executescript('CREATE TABLE IF NOT EXISTS fact(id TEXT PRIMARY KEY, version INTEGER, amount_cents INTEGER, deleted INTEGER); CREATE TABLE IF NOT EXISTS checkpoint(id INTEGER PRIMARY KEY CHECK(id=1),pos INTEGER); INSERT OR IGNORE INTO checkpoint VALUES(1,0); CREATE TABLE IF NOT EXISTS batches(start_pos INTEGER,end_pos INTEGER,source_rows INTEGER, UNIQUE(start_pos,end_pos));')
connection.commit()
finally:
connection.close()
def read_state(path):
connection = connect(path)
try:
return {'rows':[list(r) for r in connection.execute('SELECT id,version,amount_cents FROM fact WHERE deleted=0 ORDER BY id')],
'checkpoint':connection.execute('SELECT pos FROM checkpoint WHERE id=1').fetchone()[0],
'batches':connection.execute('SELECT count(*) FROM batches').fetchone()[0]}
finally:
connection.close()
def audit_source_rows(path):
connection = connect(path)
try:
return connection.execute('SELECT coalesce(sum(source_rows),0) FROM batches').fetchone()[0]
finally:
connection.close()
def apply_batch(path, events, fault=None):
connection = connect(path)
try:
old = connection.execute('SELECT pos FROM checkpoint WHERE id=1').fetchone()[0]
if any(type(r['pos']) is not int or r['pos'] <= 0 for r in events):
raise ValueError('positive integer positions required')
selected = sorted((r for r in events if r['pos'] > old), key=lambda r:r['pos'])
if not selected:
return
if [r['pos'] for r in selected] != list(range(old+1,selected[-1]['pos']+1)):
raise ValueError('source gap or duplicate position')
for r in selected:
if r['op'] not in ['upsert','delete'] or not isinstance(r['id'],str) or not r['id'] or type(r['version']) is not int or r['version'] < 1:
raise ValueError('invalid source envelope')
if r['op'] == 'upsert' and type(r.get('amount_cents')) is not int:
raise ValueError('integer cents required')
with connection:
for r in selected:
payload = (r.get('amount_cents') if r['op']=='upsert' else None, int(r['op']=='delete'))
previous = connection.execute('SELECT version,amount_cents,deleted FROM fact WHERE id=?',[r['id']]).fetchone()
if previous and previous[0] == r['version'] and tuple(previous[1:]) != payload:
raise ValueError('conflicting source version')
connection.execute('INSERT INTO fact VALUES(?,?,?,?) ON CONFLICT(id) DO UPDATE SET version=excluded.version,amount_cents=excluded.amount_cents,deleted=excluded.deleted WHERE excluded.version > fact.version',[r['id'],r['version'],*payload])
if fault:
fault()
connection.execute('UPDATE checkpoint SET pos=? WHERE id=1',[selected[-1]['pos']])
connection.execute('INSERT INTO batches VALUES(?,?,?)',[old,selected[-1]['pos'],len(selected)])
finally:
connection.close()
def inspect_example(events, fail):
path = 'capstone.db'
prepare(path)
def interrupted():
raise RuntimeError('controlled pre-commit interruption')
try:
apply_batch(path,events,interrupted if fail else None)
except RuntimeError:
pass
before = read_state(path)
apply_batch(path,events)
after = read_state(path)
apply_batch(path,events)
return {'before':before,'after':after,'replayUnchanged':after == read_state(path)}- Reasoning
- prepare creates the tables and initial zero checkpoint. apply_batch selects positions beyond that checkpoint, rejects gaps or duplicates, validates envelopes and applies version-guarded UPSERTs with retained tombstones. It rejects different payloads at one entity version. Only after data statements does it update progress and insert the batch audit within the same transaction.