Skip to content

Instantly share code, notes, and snippets.

@Directory
Created November 3, 2019 07:02
Show Gist options
  • Save Directory/39ed1f463c74bc2c785d00a5074cc279 to your computer and use it in GitHub Desktop.
Save Directory/39ed1f463c74bc2c785d00a5074cc279 to your computer and use it in GitHub Desktop.
This is an example demo of how I was able to pass an sqlite3 database in memory between threads.
import sqlite3
from threading import Thread
import time
conn = sqlite3.connect("file::memory:?cache=shared", uri=True, check_same_thread=False)
conn.execute("PRAGMA journal_mode=MEMORY")
c = conn.cursor()
# s = time.time()
c.execute('CREATE TABLE test (test TEXT, test2 TEXT)')
c.execute('INSERT INTO test VALUES (?,?)', ('test', 'yo yo',))
c.execute('SELECT * FROM test')
conn.commit()
# print(f'time taken {time.time() - s}')
print(c.fetchall(), '\n\n')
def test_func():
# s1 = time.time()
con = sqlite3.connect("file::memory:?cache=shared", uri=True)
cur = con.cursor()
cur.execute('SELECT * FROM test')
# print(f'time taken {time.time() - s1}')
print(cur.fetchall())
# alternativly, you could skip making the new connecton and curser objects and use the old one as the
# check_same_thread argument is enabled. Only do this as a last resort if the original memory connection
# can't be connected to, as it sometimes happens to me, and I think it means its outside of the memorys
# scope of reach
time.sleep(1)
while True:
Thread(target=test_func).start()
# test_func()
time.sleep(1)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment