Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import sqlite3
import threading
db = sqlite3.connect(":memory:", check_same_thread=False)
CREATE_NON_LINKED_TABLE = 'CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY AUTOINCREMENT, original, cleaned, UNIQUE(original))'
CREATE_LINKED_TABLE = 'CREATE TABLE IF NOT EXISTS %s (id INTEGER PRIMARY KEY AUTOINCREMENT, original, cleaned, study INTEGER, UNIQUE(original, study), FOREIGN KEY(study) REFERENCES studyinstanceuid(id))'
db.isolation_level = None
db.execute(CREATE_NON_LINKED_TABLE % 'studyinstanceuid')
db.execute(CREATE_LINKED_TABLE % 'patientsname')
INSERT_OTHER = 'INSERT INTO %s (original, cleaned) VALUES (?, ?)'
INSERT_LINKED = 'INSERT INTO %s (original, cleaned, study) VALUES (?, ?, ?)'
db.execute(INSERT_OTHER % 'studyinstanceuid', ('original', 'cleaned'))
db.execute(INSERT_LINKED % 'patientsname', ('original', 'cleaned', 1))
def foo():
while True:
try:
db.execute(INSERT_LINKED % 'patientsname', ('original', 'cleaned', 1))
except sqlite3.IntegrityError as ex:
pass
a = threading.Thread(target=foo)
a.start()
b = threading.Thread(target=foo)
b.start()
a.join()
b.join()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.