Skip to content
Merged
Next Next commit
bpo-45613: Dynamically set sqlite3.threadsafety
Use the compile-time selected default SQLite threaded mode to set the
DB-API 2.0 attribute 'threadsafety'

Mappings:
  - SQLITE_THREADSAFE=0 => threadsafety=0
  - SQLITE_THREADSAFE=1 => threadsafety=3
  - SQLITE_THREADSAFE=2 => threadsafety=1
  • Loading branch information
Erlend E. Aasland
Erlend E. Aasland committed Oct 26, 2021
commit bf9677c7cf697d50c81f4ddcb41b93918c01bec9
32 changes: 31 additions & 1 deletion Lib/sqlite3/dbapi2.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,37 @@

paramstyle = "qmark"

threadsafety = 1

def fetch_compile_options():
try:
cx = connect(":memory:")
res = cx.execute("select * from pragma_compile_options")
opts = {i[0] for i in res}
finally:
cx.close()
return opts

_compile_options = fetch_compile_options()
del fetch_compile_options

def _threadsafety():
if "THREADSAFE=0" in _compile_options: # SQLite single-thread mode
# Translates to DB-API threadsafety level 0, meaning threads may not
# share the module.
return 0
elif "THREADSAFE=2" in _compile_options:
# SQLITE_THREADSAFE=2 (multi-thread mode) translates to DB-API
# threadsafety level 1, meaning threads may share the module, but not
# connections.
return 1

# Default to SQLITE_THREADSAFE=1 (serialized mode), which translates to
# DB-API threadsafety level 3, meaning threads may share the module,
# connections and cursors. This is the default threaded mode of SQLite.
return 3

threadsafety = _threadsafety()
del _threadsafety

apilevel = "2.0"

Expand Down
5 changes: 3 additions & 2 deletions Lib/sqlite3/test/test_dbapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,8 +54,9 @@ def test_api_level(self):
"apilevel is %s, should be 2.0" % sqlite.apilevel)

def test_thread_safety(self):
self.assertEqual(sqlite.threadsafety, 1,
"threadsafety is %d, should be 1" % sqlite.threadsafety)
self.assertIn(sqlite.threadsafety, {0, 1, 3},
"threadsafety is %d, should be 0, 1 or 3" %
sqlite.threadsafety)

def test_param_style(self):
self.assertEqual(sqlite.paramstyle, "qmark",
Expand Down