sqlite_locking_test.py

import time
import os
from multiprocessing import Pool
import sqlite3

fname = 'sqlite_locking_test.sqlite'

def f(args):
    n, delay = args
    for i in range(n):
        db = sqlite3.connect(fname, timeout=1)
        db.execute("insert into foo values (?)", ('x'*16000, ))
        db.commit()
        db.close()
        time.sleep(delay)

if __name__ == '__main__':
    try:
        os.unlink(fname)
    except OSError: pass

    db = sqlite3.connect(fname)
    db.execute("create table if not exists foo (bar text);")
    db.close()

    concurrency = 4
    tiles_per_proc = 2000
    delay = 0.005 # time to "create" new tile

    p = Pool(concurrency)
    p.map(f, [(tiles_per_proc, delay)] * concurrency)

    db = sqlite3.connect(fname)
    print db.execute("select count(*) from foo;").next()
    db.close()