SQLite3 multi-value bulk insert PK generation consequence test

HISTORY.rst

Migrated from https://bitbucket.org/saaj/workspace/snippets/bed59d/sqlite3-multi-value-bulk-insert-pk

sqlitetest.py

#!/usr/bin/env python


import uuid
import time
import logging
import sqlite3
import threading
from multiprocessing.dummy import Pool
#from multiprocessing import Pool


local = threading.local()

def query(sql, params=()):
    if not hasattr(local, 'db'):
        local.db = sqlite3.connect('test.sqlite', timeout = 300, isolation_level = None)
    
    cursor = local.db.cursor()
    cursor.execute(sql, params)
    
    return {
      'rows': cursor.fetchall(), 
      'lastrowid': cursor.lastrowid, 
      'rowcount': cursor.rowcount}
  

def ensure_table():
    sql = '''
        CREATE TABLE IF NOT EXISTS test ( 
          test_id   INTEGER PRIMARY KEY NOT NULL,
          uuid TEST NOT NULL
        )
    '''
    return query(sql)


def test_insert(index):
    logging.debug('Inserting part %s', index)
    
    numrows = 100
    sql = 'INSERT INTO test(uuid) VALUES {}'.format(',\n'.join(['(?)'] * numrows))
    # uuid1() should produce chronologically orderable sequence
    uuids = [str(uuid.uuid1()) for _ in range(numrows)]
    cursor_data = query(sql, uuids)
    
    assert cursor_data['rowcount'] == numrows 
    expected_ids = list(range(
      cursor_data['lastrowid'] - cursor_data['rowcount'] + 1, 
      cursor_data['lastrowid'] + 1))
    
    sql = '''
        SELECT test_id
        FROM test
        WHERE uuid IN({})
        ORDER BY uuid
    '''.format(','.join(['?'] * numrows))
    cursor_data = query(sql, uuids)
    
    assert [r[0] for r in cursor_data['rows']] == expected_ids


if __name__ == '__main__':
    logging.basicConfig(level = logging.DEBUG)
    pool = Pool(20)
    pool.apply(ensure_table)
    pool.map(test_insert, range(1000), chunksize = 1)
    pool.close()
    pool.join()