diff --git a/Lib/test/test_sqlite3/test_dbapi.py b/Lib/test/test_sqlite3/test_dbapi.py
index bbb7c1f8b615938..e57b44fc7efd429 100644
--- a/Lib/test/test_sqlite3/test_dbapi.py
+++ b/Lib/test/test_sqlite3/test_dbapi.py
@@ -26,6 +26,7 @@
 import sqlite3 as sqlite
 import subprocess
 import sys
+import textwrap
 import threading
 import unittest
 import urllib.parse
@@ -1734,6 +1735,118 @@ def run(con, err):
             t.join()
             self.assertEqual(len(err), 0, "\n".join(err))
 
+    @requires_subprocess()
+    @unittest.skipUnless(sqlite.threadsafety == 3,
+                         "requires SQLite in serialized threading mode")
+    def test_shared_connection_callback_deadlock(self):
+        # Threads sharing a connection must not deadlock when SQLite calls
+        # back into Python.  A deadlock would hang the whole process, so run
+        # this in a child process.
+        SCRIPT = textwrap.dedent("""
+            import itertools
+            import sqlite3
+            import sys
+            import threading
+
+            sys.setswitchinterval(0.00001)
+
+            class Count:
+                def __init__(self):
+                    self.n = 0
+                def step(self, x):
+                    self.n += 1
+                def finalize(self):
+                    return self.n
+
+            def connect():
+                cx = sqlite3.connect(":memory:", check_same_thread=False)
+                cx.execute("create table t(x)")
+                cx.executemany("insert into t values(?)",
+                               ((str(i),) for i in range(50)))
+                return cx
+
+            def query(cx, sql, n, errors):
+                # Use different SQL in each thread and for each iteration:
+                # the threads must not share a cached statement, and the
+                # authorizer is only called when a statement is prepared.
+                try:
+                    for i in range(n, n + 100):
+                        params = ("", 100) if i % 2 else [None, 100]
+                        cx.execute(f"{sql} limit ?+{i}", params).fetchall()
+                except Exception as exc:
+                    errors.append(exc)
+
+            def run(name, sql, setup, concurrent=None):
+                print(name, flush=True)
+                cx = connect()
+                setup(cx)
+                errors = []
+                threads = [threading.Thread(target=query,
+                                            args=(cx, sql, 1000, errors))]
+                if concurrent is None:
+                    threads.append(threading.Thread(
+                        target=query, args=(cx, sql, 2000, errors)))
+                for t in threads:
+                    t.start()
+                while concurrent is not None and threads[0].is_alive():
+                    concurrent(cx)
+                for t in threads:
+                    t.join()
+                cx.close()
+                if errors:
+                    raise errors[0]
+
+            def function(cx):
+                cx.create_function("f", 1, lambda x: x)
+            def collation(cx):
+                cx.create_collation("c", lambda a, b: (a > b) - (a < b))
+
+            run("function", "select f(x) from t where x is not ?", function)
+            run("aggregate", "select agg(x) from t where x is not ? group by x",
+                lambda cx: cx.create_aggregate("agg", 1, Count))
+            run("collation",
+                "select x from t where x is not ? order by x collate c",
+                collation)
+            run("authorizer", "select x from t where x is not ?",
+                lambda cx: cx.set_authorizer(lambda *a: sqlite3.SQLITE_OK))
+            run("progress handler", "select x from t where x is not ?",
+                lambda cx: cx.set_progress_handler(lambda: 0, 1))
+            run("trace callback", "select x from t where x is not ?",
+                lambda cx: cx.set_trace_callback(lambda stmt: None))
+
+            # Register callbacks while another thread is in a callback.
+            # Use new names: existing functions and collations cannot be
+            # modified while statements are running.
+            sql = "select f(x) from t where x is not ?"
+            names = map("name{}".format, itertools.count())
+            run("create_function", sql, function,
+                lambda cx: cx.create_function(next(names), 0, lambda: None))
+            run("create_aggregate", sql, function,
+                lambda cx: cx.create_aggregate(next(names), 1, Count))
+            run("create_collation", sql, function,
+                lambda cx: cx.create_collation(next(names), lambda a, b: 0))
+            run("set_authorizer", sql, function,
+                lambda cx: cx.set_authorizer(None))
+            run("set_progress_handler", sql, function,
+                lambda cx: cx.set_progress_handler(None, 0))
+            run("set_trace_callback", sql, function,
+                lambda cx: cx.set_trace_callback(None))
+            run("setconfig", sql, function,
+                lambda cx: cx.setconfig(sqlite3.SQLITE_DBCONFIG_ENABLE_FKEY))
+            run("getters", sql, function,
+                lambda cx: (cx.in_transaction, cx.total_changes,
+                            cx.getlimit(sqlite3.SQLITE_LIMIT_LENGTH)))
+            print("done")
+        """)
+        try:
+            proc = subprocess.run([sys.executable, "-c", SCRIPT],
+                                  capture_output=True, text=True,
+                                  timeout=SHORT_TIMEOUT)
+        except subprocess.TimeoutExpired as exc:
+            self.fail(f"deadlock; output: {exc.stdout!r}")
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertEndsWith(proc.stdout, "done\n")
+
 
 class ConstructorTests(unittest.TestCase):
     def test_date(self):
diff --git a/Misc/NEWS.d/next/Library/2026-10-22-06-39-13.gh-issue-161065.q7GdMx.rst b/Misc/NEWS.d/next/Library/2026-10-22-06-39-13.gh-issue-161065.q7GdMx.rst
new file mode 100644
index 000000000000000..9e566e386ba74de
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-10-22-06-39-13.gh-issue-161065.q7GdMx.rst
@@ -0,0 +1,4 @@
+Fix a deadlock in :mod:`sqlite3` when a connection created with
+``check_same_thread=False`` is used from several threads at once and has a
+Python callback registered, such as a user-defined function, aggregate,
+collation, authorizer, progress handler or trace callback.
diff --git a/Modules/_sqlite/connection.c b/Modules/_sqlite/connection.c
index f8bd3a8df35b668..a9cb774eac0b205 100644
--- a/Modules/_sqlite/connection.c
+++ b/Modules/_sqlite/connection.c
@@ -434,6 +434,7 @@ remove_callbacks(sqlite3 *db)
     /* None of these APIs can fail, as long as they are given a valid
      * database pointer. */
     assert(db != NULL);
+    pysqlite_db_mutex_enter(db);
     int rc = sqlite3_trace_v2(db, SQLITE_TRACE_STMT, 0, 0);
     assert(rc == SQLITE_OK), (void)rc;
 
@@ -441,6 +442,7 @@ remove_callbacks(sqlite3 *db)
 
     rc = sqlite3_set_authorizer(db, NULL, NULL);
     assert(rc == SQLITE_OK), (void)rc;
+    pysqlite_db_mutex_leave(db);
 }
 
 static int
@@ -452,7 +454,7 @@ connection_close(pysqlite_Connection *self)
 
     int rc = 0;
     if (self->autocommit == AUTOCOMMIT_DISABLED &&
-        !sqlite3_get_autocommit(self->db))
+        !pysqlite_get_autocommit(self->db))
     {
         if (connection_exec_stmt(self, "ROLLBACK") < 0) {
             rc = -1;
@@ -757,7 +759,7 @@ pysqlite_connection_commit_impl(pysqlite_Connection *self)
     }
 
     if (self->autocommit == AUTOCOMMIT_LEGACY) {
-        if (!sqlite3_get_autocommit(self->db)) {
+        if (!pysqlite_get_autocommit(self->db)) {
             if (connection_exec_stmt(self, "COMMIT") < 0) {
                 return NULL;
             }
@@ -791,7 +793,7 @@ pysqlite_connection_rollback_impl(pysqlite_Connection *self)
     }
 
     if (self->autocommit == AUTOCOMMIT_LEGACY) {
-        if (!sqlite3_get_autocommit(self->db)) {
+        if (!pysqlite_get_autocommit(self->db)) {
             if (connection_exec_stmt(self, "ROLLBACK") < 0) {
                 return NULL;
             }
@@ -1182,7 +1184,7 @@ destructor_callback(void *ctx)
 static int
 check_num_params(pysqlite_Connection *self, const int n, const char *name)
 {
-    int limit = sqlite3_limit(self->db, SQLITE_LIMIT_FUNCTION_ARG, -1);
+    int limit = pysqlite_limit(self->db, SQLITE_LIMIT_FUNCTION_ARG, -1);
     assert(limit >= 0);
     if (n < -1 || n > limit) {
         PyErr_Format(self->ProgrammingError,
@@ -1231,11 +1233,13 @@ pysqlite_connection_create_function_impl(pysqlite_Connection *self,
     if (ctx == NULL) {
         return NULL;
     }
+    pysqlite_db_mutex_enter(self->db);
     rc = sqlite3_create_function_v2(self->db, name, narg, flags, ctx,
                                     func_callback,
                                     NULL,
                                     NULL,
                                     &destructor_callback);  // will free 'ctx'
+    pysqlite_db_mutex_leave(self->db);
 
     if (rc != SQLITE_OK) {
         /* Workaround for SQLite bug: no error code or string is available here */
@@ -1376,14 +1380,17 @@ create_window_function_impl(pysqlite_Connection *self, PyTypeObject *cls,
     int flags = SQLITE_UTF8;
     int rc;
     if (Py_IsNone(aggregate_class)) {
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_create_window_function(self->db, name, num_params, flags,
                                             0, 0, 0, 0, 0, 0);
+        pysqlite_db_mutex_leave(self->db);
     }
     else {
         callback_context *ctx = create_callback_context(cls, aggregate_class);
         if (ctx == NULL) {
             return NULL;
         }
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_create_window_function(self->db, name, num_params, flags,
                                             ctx,
                                             &step_callback,
@@ -1391,6 +1398,7 @@ create_window_function_impl(pysqlite_Connection *self, PyTypeObject *cls,
                                             &value_callback,
                                             &inverse_callback,
                                             &destructor_callback);
+        pysqlite_db_mutex_leave(self->db);
     }
 
     if (rc != SQLITE_OK) {
@@ -1435,11 +1443,13 @@ pysqlite_connection_create_aggregate_impl(pysqlite_Connection *self,
     if (ctx == NULL) {
         return NULL;
     }
+    pysqlite_db_mutex_enter(self->db);
     rc = sqlite3_create_function_v2(self->db, name, n_arg, SQLITE_UTF8, ctx,
                                     0,
                                     &step_callback,
                                     &final_callback,
                                     &destructor_callback); // will free 'ctx'
+    pysqlite_db_mutex_leave(self->db);
     if (rc != SQLITE_OK) {
         /* Workaround for SQLite bug: no error code or string is available here */
         PyErr_SetString(self->OperationalError, "Error creating aggregate");
@@ -1593,7 +1603,9 @@ pysqlite_connection_set_authorizer_impl(pysqlite_Connection *self,
 
     int rc;
     if (callable == Py_None) {
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_set_authorizer(self->db, NULL, NULL);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->authorizer_ctx, NULL);
     }
     else {
@@ -1601,7 +1613,9 @@ pysqlite_connection_set_authorizer_impl(pysqlite_Connection *self,
         if (ctx == NULL) {
             return NULL;
         }
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_set_authorizer(self->db, authorizer_callback, ctx);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->authorizer_ctx, ctx);
     }
     if (rc != SQLITE_OK) {
@@ -1644,7 +1658,9 @@ pysqlite_connection_set_progress_handler_impl(pysqlite_Connection *self,
 
     if (callable == Py_None) {
         /* None clears the progress handler previously set */
+        pysqlite_db_mutex_enter(self->db);
         sqlite3_progress_handler(self->db, 0, 0, (void*)0);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->progress_ctx, NULL);
     }
     else {
@@ -1652,7 +1668,9 @@ pysqlite_connection_set_progress_handler_impl(pysqlite_Connection *self,
         if (ctx == NULL) {
             return NULL;
         }
+        pysqlite_db_mutex_enter(self->db);
         sqlite3_progress_handler(self->db, n, progress_callback, ctx);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->progress_ctx, ctx);
     }
     Py_RETURN_NONE;
@@ -1687,7 +1705,9 @@ pysqlite_connection_set_trace_callback_impl(pysqlite_Connection *self,
          * - https://sqlite.org/c3ref/c_trace.html
          * - https://sqlite.org/c3ref/trace_v2.html
          */
+        pysqlite_db_mutex_enter(self->db);
         sqlite3_trace_v2(self->db, SQLITE_TRACE_STMT, 0, 0);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->trace_ctx, NULL);
     }
     else {
@@ -1695,7 +1715,9 @@ pysqlite_connection_set_trace_callback_impl(pysqlite_Connection *self,
         if (ctx == NULL) {
             return NULL;
         }
+        pysqlite_db_mutex_enter(self->db);
         sqlite3_trace_v2(self->db, SQLITE_TRACE_STMT, trace_callback, ctx);
+        pysqlite_db_mutex_leave(self->db);
         set_callback_context(&self->trace_ctx, ctx);
     }
 
@@ -1728,7 +1750,9 @@ pysqlite_connection_enable_load_extension_impl(pysqlite_Connection *self,
         return NULL;
     }
 
+    pysqlite_db_mutex_enter(self->db);
     rc = sqlite3_enable_load_extension(self->db, onoff);
+    pysqlite_db_mutex_leave(self->db);
 
     if (rc != SQLITE_OK) {
         PyErr_SetString(self->OperationalError,
@@ -1767,7 +1791,9 @@ pysqlite_connection_load_extension_impl(pysqlite_Connection *self,
         return NULL;
     }
 
+    pysqlite_db_mutex_enter(self->db);
     rc = sqlite3_load_extension(self->db, extension_name, entrypoint, &errmsg);
+    pysqlite_db_mutex_leave(self->db);
     if (rc != 0) {
         PyErr_SetString(self->OperationalError, errmsg);
         return NULL;
@@ -1812,7 +1838,10 @@ pysqlite_connection_get_total_changes(PyObject *op, void *Py_UNUSED(closure))
     if (!pysqlite_check_connection(self)) {
         return NULL;
     }
-    return PyLong_FromLong(sqlite3_total_changes(self->db));
+    pysqlite_db_mutex_enter(self->db);
+    int changes = sqlite3_total_changes(self->db);
+    pysqlite_db_mutex_leave(self->db);
+    return PyLong_FromLong(changes);
 }
 
 static PyObject *
@@ -1822,7 +1851,7 @@ pysqlite_connection_get_in_transaction(PyObject *op, void *Py_UNUSED(closure))
     if (!pysqlite_check_connection(self)) {
         return NULL;
     }
-    if (!sqlite3_get_autocommit(self->db)) {
+    if (!pysqlite_get_autocommit(self->db)) {
         Py_RETURN_TRUE;
     }
     Py_RETURN_FALSE;
@@ -2232,8 +2261,10 @@ pysqlite_connection_create_collation_impl(pysqlite_Connection *self,
     int rc;
     int flags = SQLITE_UTF8;
     if (callable == Py_None) {
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_create_collation_v2(self->db, name, flags,
                                          NULL, NULL, NULL);
+        pysqlite_db_mutex_leave(self->db);
     }
     else {
         if (!PyCallable_Check(callable)) {
@@ -2244,9 +2275,11 @@ pysqlite_connection_create_collation_impl(pysqlite_Connection *self,
         if (ctx == NULL) {
             return NULL;
         }
+        pysqlite_db_mutex_enter(self->db);
         rc = sqlite3_create_collation_v2(self->db, name, flags, ctx,
                                          &collation_callback,
                                          &destructor_callback);
+        pysqlite_db_mutex_leave(self->db);
     }
 
     if (rc != SQLITE_OK) {
@@ -2478,7 +2511,7 @@ setlimit_impl(pysqlite_Connection *self, int category, int limit)
         return NULL;
     }
 
-    int old_limit = sqlite3_limit(self->db, category, limit);
+    int old_limit = pysqlite_limit(self->db, category, limit);
     if (old_limit < 0) {
         PyErr_SetString(self->ProgrammingError, "'category' is out of bounds");
         return NULL;
@@ -2570,7 +2603,9 @@ setconfig_impl(pysqlite_Connection *self, int op, int enable)
     }
 
     int actual;
+    pysqlite_db_mutex_enter(self->db);
     int rc = sqlite3_db_config(self->db, op, enable, &actual);
+    pysqlite_db_mutex_leave(self->db);
     if (rc != SQLITE_OK) {
         set_error_from_db(self->state, self->db);
         return NULL;
@@ -2605,7 +2640,9 @@ getconfig_impl(pysqlite_Connection *self, int op)
     }
 
     int current;
+    pysqlite_db_mutex_enter(self->db);
     int rc = sqlite3_db_config(self->db, op, -1, &current);
+    pysqlite_db_mutex_leave(self->db);
     if (rc != SQLITE_OK) {
         set_error_from_db(self->state, self->db);
         return -1;
@@ -2645,14 +2682,14 @@ set_autocommit(PyObject *op, PyObject *val, void *Py_UNUSED(closure))
         return -1;
     }
     if (self->autocommit == AUTOCOMMIT_ENABLED) {
-        if (!sqlite3_get_autocommit(self->db)) {
+        if (!pysqlite_get_autocommit(self->db)) {
             if (connection_exec_stmt(self, "COMMIT") < 0) {
                 return -1;
             }
         }
     }
     else if (self->autocommit == AUTOCOMMIT_DISABLED) {
-        if (sqlite3_get_autocommit(self->db)) {
+        if (pysqlite_get_autocommit(self->db)) {
             if (connection_exec_stmt(self, "BEGIN") < 0) {
                 return -1;
             }
diff --git a/Modules/_sqlite/cursor.c b/Modules/_sqlite/cursor.c
index ad8c434c1b2fb8f..2c116ab4aa7ae37 100644
--- a/Modules/_sqlite/cursor.c
+++ b/Modules/_sqlite/cursor.c
@@ -246,6 +246,8 @@ pysqlite_build_row_cast_map(pysqlite_Cursor* self)
         return -1;
     }
 
+    sqlite3 *db = self->connection->db;
+    pysqlite_db_mutex_enter(db);
     for (i = 0; i < sqlite3_column_count(self->statement->st); i++) {
         converter = NULL;
 
@@ -253,8 +255,7 @@ pysqlite_build_row_cast_map(pysqlite_Cursor* self)
             const char *colname = sqlite3_column_name(self->statement->st, i);
             if (colname == NULL) {
                 PyErr_NoMemory();
-                Py_CLEAR(self->row_cast_map);
-                return -1;
+                goto error;
             }
             const char *type_start = NULL;
             for (pos = colname; *pos != 0; pos++) {
@@ -266,8 +267,7 @@ pysqlite_build_row_cast_map(pysqlite_Cursor* self)
                     converter = _pysqlite_get_converter(state, type_start,
                                                         pos - type_start);
                     if (!converter && PyErr_Occurred()) {
-                        Py_CLEAR(self->row_cast_map);
-                        return -1;
+                        goto error;
                     }
                     break;
                 }
@@ -287,8 +287,7 @@ pysqlite_build_row_cast_map(pysqlite_Cursor* self)
                         converter = _pysqlite_get_converter(state, decltype,
                                                             pos - decltype);
                         if (!converter && PyErr_Occurred()) {
-                            Py_CLEAR(self->row_cast_map);
-                            return -1;
+                            goto error;
                         }
                         break;
                     }
@@ -300,12 +299,17 @@ pysqlite_build_row_cast_map(pysqlite_Cursor* self)
                                converter ? converter : Py_None);
         Py_XDECREF(converter);
         if (rc != 0) {
-            Py_CLEAR(self->row_cast_map);
-            return -1;
+            goto error;
         }
     }
+    pysqlite_db_mutex_leave(db);
 
     return 0;
+
+error:
+    pysqlite_db_mutex_leave(db);
+    Py_CLEAR(self->row_cast_map);
+    return -1;
 }
 
 static PyObject *
@@ -359,6 +363,10 @@ _pysqlite_fetch_one_row(pysqlite_Cursor* self)
         return NULL;
 
     sqlite3 *db = self->connection->db;
+    // Hold the connection mutex once for the whole row instead of letting each
+    // sqlite3_column_*() call take it.  Converters and the text factory run
+    // Python code, so the mutex is released around those calls.
+    pysqlite_db_mutex_enter(db);
     for (i = 0; i < numcols; i++) {
         if (self->connection->detect_types
                 && self->row_cast_map != NULL
@@ -391,13 +399,13 @@ _pysqlite_fetch_one_row(pysqlite_Cursor* self)
                 if (item == NULL) {
                     goto error;
                 }
+                pysqlite_db_mutex_leave(db);
                 converted = PyObject_CallOneArg(converter, item);
                 Py_DECREF(item);
+                pysqlite_db_mutex_enter(db);
             }
         } else {
-            Py_BEGIN_ALLOW_THREADS
             coltype = sqlite3_column_type(self->statement->st, i);
-            Py_END_ALLOW_THREADS
             if (coltype == SQLITE_NULL) {
                 converted = Py_NewRef(Py_None);
             } else if (coltype == SQLITE_INTEGER) {
@@ -443,7 +451,15 @@ _pysqlite_fetch_one_row(pysqlite_Cursor* self)
                 } else if (factory == (PyObject*)&PyByteArray_Type) {
                     converted = PyByteArray_FromStringAndSize(text, nbytes);
                 } else {
-                    converted = PyObject_CallFunction(factory, "y#", text, nbytes);
+                    PyObject *item = PyBytes_FromStringAndSize(text, nbytes);
+                    if (item == NULL) {
+                        Py_DECREF(factory);
+                        goto error;
+                    }
+                    pysqlite_db_mutex_leave(db);
+                    converted = PyObject_CallOneArg(factory, item);
+                    Py_DECREF(item);
+                    pysqlite_db_mutex_enter(db);
                 }
                 Py_DECREF(factory);
             } else {
@@ -464,13 +480,17 @@ _pysqlite_fetch_one_row(pysqlite_Cursor* self)
         }
         PyTuple_SET_ITEM(row, i, converted);
     }
+    pysqlite_db_mutex_leave(db);
 
-    if (PyErr_Occurred())
-        goto error;
+    if (PyErr_Occurred()) {
+        Py_DECREF(row);
+        return NULL;
+    }
 
     return row;
 
 error:
+    pysqlite_db_mutex_leave(db);
     Py_DECREF(row);
     return NULL;
 }
@@ -554,8 +574,8 @@ stmt_step(sqlite3_stmt *statement)
 }
 
 static int
-bind_param(pysqlite_state *state, pysqlite_Statement *self, int pos,
-           PyObject *parameter)
+bind_param(pysqlite_state *state, pysqlite_Statement *self,
+           sqlite3_mutex *mutex, int pos, PyObject *parameter)
 {
     int rc = SQLITE_OK;
     const char *string;
@@ -563,7 +583,9 @@ bind_param(pysqlite_state *state, pysqlite_Statement *self, int pos,
     parameter_type paramtype;
 
     if (parameter == Py_None) {
+        pysqlite_mutex_enter(mutex);
         rc = sqlite3_bind_null(self->st, pos);
+        pysqlite_mutex_leave(mutex);
         goto final;
     }
 
@@ -588,16 +610,22 @@ bind_param(pysqlite_state *state, pysqlite_Statement *self, int pos,
     switch (paramtype) {
         case TYPE_LONG: {
             sqlite_int64 value = _pysqlite_long_as_int64(parameter);
-            rc = (value == -1 && PyErr_Occurred())
-                ? SQLITE_ERROR
-                : sqlite3_bind_int64(self->st, pos, value);
+            if (value == -1 && PyErr_Occurred()) {
+                return SQLITE_ERROR;
+            }
+            pysqlite_mutex_enter(mutex);
+            rc = sqlite3_bind_int64(self->st, pos, value);
+            pysqlite_mutex_leave(mutex);
             break;
         }
         case TYPE_FLOAT: {
             double value = PyFloat_AsDouble(parameter);
-            rc = (value == -1 && PyErr_Occurred())
-                ? SQLITE_ERROR
-                : sqlite3_bind_double(self->st, pos, value);
+            if (value == -1 && PyErr_Occurred()) {
+                return SQLITE_ERROR;
+            }
+            pysqlite_mutex_enter(mutex);
+            rc = sqlite3_bind_double(self->st, pos, value);
+            pysqlite_mutex_leave(mutex);
             break;
         }
         case TYPE_UNICODE:
@@ -610,7 +638,9 @@ bind_param(pysqlite_state *state, pysqlite_Statement *self, int pos,
                                 "string longer than INT_MAX bytes");
                 return SQLITE_ERROR;
             }
+            pysqlite_mutex_enter(mutex);
             rc = sqlite3_bind_text(self->st, pos, string, (int)buflen, SQLITE_TRANSIENT);
+            pysqlite_mutex_leave(mutex);
             break;
         case TYPE_BUFFER: {
             Py_buffer view;
@@ -623,7 +653,9 @@ bind_param(pysqlite_state *state, pysqlite_Statement *self, int pos,
                 PyBuffer_Release(&view);
                 return SQLITE_ERROR;
             }
+            pysqlite_mutex_enter(mutex);
             rc = sqlite3_bind_blob(self->st, pos, view.buf, (int)view.len, SQLITE_TRANSIENT);
+            pysqlite_mutex_leave(mutex);
             PyBuffer_Release(&view);
             break;
         }
@@ -654,9 +686,25 @@ need_adapt(pysqlite_state *state, PyObject *obj)
     }
 }
 
+/* Return 1 if binding the parameters cannot call back into Python code,
+ * that is if they are a tuple of objects that need no adaptation. */
+static inline int
+simple_parameters(pysqlite_state *state, PyObject *parameters)
+{
+    if (!PyTuple_CheckExact(parameters)) {
+        return 0;
+    }
+    for (Py_ssize_t i = 0; i < PyTuple_GET_SIZE(parameters); i++) {
+        if (need_adapt(state, PyTuple_GET_ITEM(parameters, i))) {
+            return 0;
+        }
+    }
+    return 1;
+}
+
 static void
 bind_parameters(pysqlite_state *state, pysqlite_Statement *self,
-                PyObject *parameters)
+                sqlite3_mutex *mutex, PyObject *parameters)
 {
     PyObject* current_param;
     PyObject* adapted;
@@ -666,9 +714,7 @@ bind_parameters(pysqlite_state *state, pysqlite_Statement *self,
     int num_params_needed;
     Py_ssize_t num_params;
 
-    Py_BEGIN_ALLOW_THREADS
     num_params_needed = sqlite3_bind_parameter_count(self->st);
-    Py_END_ALLOW_THREADS
 
     if (PyTuple_CheckExact(parameters) || PyList_CheckExact(parameters) || (!PyDict_Check(parameters) && PySequence_Check(parameters))) {
         /* parameters passed as sequence */
@@ -725,7 +771,7 @@ bind_parameters(pysqlite_state *state, pysqlite_Statement *self,
                 }
             }
 
-            rc = bind_param(state, self, i + 1, adapted);
+            rc = bind_param(state, self, mutex, i + 1, adapted);
             Py_DECREF(adapted);
 
             if (rc != SQLITE_OK) {
@@ -777,7 +823,7 @@ bind_parameters(pysqlite_state *state, pysqlite_Statement *self,
                 }
             }
 
-            rc = bind_param(state, self, i, adapted);
+            rc = bind_param(state, self, mutex, i, adapted);
             Py_DECREF(adapted);
 
             if (rc != SQLITE_OK) {
@@ -866,6 +912,7 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
     }
 
     pysqlite_state *state = self->connection->state;
+    sqlite3 *db = self->connection->db;
     if (multiple && sqlite3_stmt_readonly(self->statement->st)) {
         PyErr_SetString(state->ProgrammingError,
                         "executemany() can only execute DML statements.");
@@ -890,7 +937,7 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
     if (self->connection->autocommit == AUTOCOMMIT_LEGACY
         && self->connection->isolation_level
         && self->statement->is_dml
-        && sqlite3_get_autocommit(self->connection->db))
+        && pysqlite_get_autocommit(self->connection->db))
     {
         if (begin_transaction(self->connection) < 0) {
             goto error;
@@ -904,7 +951,18 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
             break;
         }
 
-        bind_parameters(state, self->statement, parameters);
+        // Holding the mutex once for all parameters is cheaper, but it must
+        // not be held while an adapter runs Python code, so parameters that
+        // may need one take it around each sqlite3_bind_*() call instead.
+        sqlite3_mutex *mutex = sqlite3_db_mutex(db);
+        if (simple_parameters(state, parameters)) {
+            pysqlite_mutex_enter(mutex);
+            bind_parameters(state, self->statement, NULL, parameters);
+            pysqlite_mutex_leave(mutex);
+        }
+        else {
+            bind_parameters(state, self->statement, mutex, parameters);
+        }
         if (PyErr_Occurred()) {
             goto error;
         }
@@ -940,12 +998,15 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
             }
             for (i = 0; i < numcols; i++) {
                 const char *colname;
+                pysqlite_db_mutex_enter(db);
                 colname = sqlite3_column_name(self->statement->st, i);
                 if (colname == NULL) {
+                    pysqlite_db_mutex_leave(db);
                     PyErr_NoMemory();
                     goto error;
                 }
                 column_name = _pysqlite_build_column_name(self, colname);
+                pysqlite_db_mutex_leave(db);
                 if (column_name == NULL) {
                     goto error;
                 }
@@ -964,7 +1025,9 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
 
         if (rc == SQLITE_DONE) {
             if (self->statement->is_dml) {
-                self->rowcount += (long)sqlite3_changes(self->connection->db);
+                pysqlite_db_mutex_enter(db);
+                self->rowcount += (long)sqlite3_changes(db);
+                pysqlite_db_mutex_leave(db);
             }
             if (stmt_reset(self->statement) != SQLITE_OK) {
                 goto reset_failure;
@@ -998,7 +1061,9 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
     if (PyErr_Occurred()) {
         if (self->statement) {
             sqlite3 *db = sqlite3_db_handle(self->statement->st);
+            pysqlite_db_mutex_enter(db);
             int sqlite3_state = sqlite3_errcode(db);
+            pysqlite_db_mutex_leave(db);
             // stmt_reset() may return a previously set exception,
             // either triggered because of Python or sqlite3.
             rc = stmt_reset(self->statement);
@@ -1085,8 +1150,8 @@ pysqlite_cursor_executescript_impl(pysqlite_Cursor *self,
     }
 
     size_t sql_len = strlen(sql_script);
-    int max_length = sqlite3_limit(self->connection->db,
-                                   SQLITE_LIMIT_SQL_LENGTH, -1);
+    int max_length = pysqlite_limit(self->connection->db,
+                                    SQLITE_LIMIT_SQL_LENGTH, -1);
     if (sql_len > (unsigned)max_length) {
         PyErr_SetString(self->connection->DataError,
                         "query string is too large");
@@ -1096,7 +1161,7 @@ pysqlite_cursor_executescript_impl(pysqlite_Cursor *self,
     // Commit if needed
     sqlite3 *db = self->connection->db;
     if (self->connection->autocommit == AUTOCOMMIT_LEGACY
-        && !sqlite3_get_autocommit(db))
+        && !pysqlite_get_autocommit(db))
     {
         int rc = SQLITE_OK;
 
@@ -1168,7 +1233,10 @@ pysqlite_cursor_iternext(PyObject *op)
     int rc = stmt_step(stmt);
     if (rc == SQLITE_DONE) {
         if (self->statement->is_dml) {
-            self->rowcount = (long)sqlite3_changes(self->connection->db);
+            sqlite3 *db = self->connection->db;
+            pysqlite_db_mutex_enter(db);
+            self->rowcount = (long)sqlite3_changes(db);
+            pysqlite_db_mutex_leave(db);
         }
         rc = stmt_reset(self->statement);
         Py_CLEAR(self->statement);
diff --git a/Modules/_sqlite/statement.c b/Modules/_sqlite/statement.c
index f31c699482f6014..88ed99f7b0ad9bb 100644
--- a/Modules/_sqlite/statement.c
+++ b/Modules/_sqlite/statement.c
@@ -49,7 +49,7 @@ pysqlite_statement_create(pysqlite_Connection *connection, PyObject *sql)
     }
 
     sqlite3 *db = connection->db;
-    int max_length = sqlite3_limit(db, SQLITE_LIMIT_SQL_LENGTH, -1);
+    int max_length = pysqlite_limit(db, SQLITE_LIMIT_SQL_LENGTH, -1);
     if (size > max_length) {
         PyErr_SetString(connection->DataError,
                         "query string is too large");
@@ -104,7 +104,10 @@ pysqlite_statement_create(pysqlite_Connection *connection, PyObject *sql)
 
 error:
     assert(PyErr_Occurred());
-    if (sqlite3_finalize(stmt) != SQLITE_OK) {
+    pysqlite_db_mutex_enter(db);
+    rc = sqlite3_finalize(stmt);
+    pysqlite_db_mutex_leave(db);
+    if (rc != SQLITE_OK) {
         PyObject *exc = PyErr_GetRaisedException();
         PyErr_SetString(connection->InternalError, "cannot finalize statement");
         _PyErr_ChainExceptions1(exc);
diff --git a/Modules/_sqlite/util.c b/Modules/_sqlite/util.c
index 177e0f668a0c59e..346b162cc6420a6 100644
--- a/Modules/_sqlite/util.c
+++ b/Modules/_sqlite/util.c
@@ -28,6 +28,7 @@
 #include "module.h"
 #include "pycore_long.h"          // _PyLong_AsByteArray()
 #include "connection.h"
+#include "util.h"
 
 // Returns non-NULL if a new exception should be raised
 static PyObject *
@@ -72,10 +73,12 @@ get_exception_class(pysqlite_state *state, int errorcode)
 }
 
 static void
-raise_exception(PyObject *type, int errcode, const char *errmsg)
+raise_exception(PyObject *type, int errcode, PyObject *errmsg)
 {
     PyObject *exc = NULL;
-    PyObject *args[] = { PyUnicode_FromString(errmsg), };
+    // errmsg is consumed.  NULL means creating it failed and an exception is
+    // already set.
+    PyObject *args[] = { errmsg, };
     if (args[0] == NULL) {
         goto exit;
     }
@@ -129,7 +132,7 @@ set_error_from_code(pysqlite_state *state, int code)
 
     const char *errmsg = sqlite3_errstr(code);
     assert(errmsg != NULL);
-    raise_exception(exc_class, code, errmsg);
+    raise_exception(exc_class, code, PyUnicode_FromString(errmsg));
 }
 
 /**
@@ -138,17 +141,20 @@ set_error_from_code(pysqlite_state *state, int code)
 int
 set_error_from_db(pysqlite_state *state, sqlite3 *db)
 {
+    pysqlite_db_mutex_enter(db);
     int errorcode = sqlite3_errcode(db);
     PyObject *exc_class = get_exception_class(state, errorcode);
     if (exc_class == NULL) {
         // No new exception need be raised.
+        pysqlite_db_mutex_leave(db);
         return SQLITE_OK;
     }
 
     /* Create and set the exception. */
     int extended_errcode = sqlite3_extended_errcode(db);
     // sqlite3_errmsg() always returns an UTF-8 encoded message
-    const char *errmsg = sqlite3_errmsg(db);
+    PyObject *errmsg = PyUnicode_FromString(sqlite3_errmsg(db));
+    pysqlite_db_mutex_leave(db);
     raise_exception(exc_class, extended_errcode, errmsg);
     return errorcode;
 }
diff --git a/Modules/_sqlite/util.h b/Modules/_sqlite/util.h
index d53aedd66b615fb..0830d9f8d9b5fed 100644
--- a/Modules/_sqlite/util.h
+++ b/Modules/_sqlite/util.h
@@ -61,4 +61,73 @@ pysqlite_set_ref(PyObject *owner, PyObject **field, PyObject *value)
     Py_XDECREF(old);
 }
 
+/* SQLite calls the user-defined callbacks, which acquire the GIL, with the
+ * database connection mutex held.  To keep that lock order, a thread must
+ * never block on the connection mutex while holding the GIL: enter the mutex
+ * with these helpers around SQLite API calls that take it and are made with
+ * the GIL held.  The mutex is recursive, and it is NULL unless SQLite is in
+ * serialized threading mode.  Newer SQLite versions take the mutex in more
+ * calls, including getters such as sqlite3_get_autocommit() and
+ * sqlite3_limit(), so assume that any call on a connection or a statement
+ * takes it unless the SQLite documentation says otherwise.
+ *
+ * Do not call back into Python code while holding the mutex: it could wait
+ * for another thread that is blocked on the same mutex. */
+static inline void
+pysqlite_mutex_enter(sqlite3_mutex *mutex)
+{
+#ifdef PY_SQLITE_HAVE_MUTEX
+    if (mutex != NULL && sqlite3_mutex_try(mutex) != SQLITE_OK) {
+        Py_BEGIN_ALLOW_THREADS
+        sqlite3_mutex_enter(mutex);
+        Py_END_ALLOW_THREADS
+    }
+#else
+    /* An SQLite built without mutexes cannot be in serialized mode. */
+    (void)mutex;
+#endif
+}
+
+static inline void
+pysqlite_mutex_leave(sqlite3_mutex *mutex)
+{
+#ifdef PY_SQLITE_HAVE_MUTEX
+    if (mutex != NULL) {
+        sqlite3_mutex_leave(mutex);
+    }
+#else
+    (void)mutex;
+#endif
+}
+
+static inline void
+pysqlite_db_mutex_enter(sqlite3 *db)
+{
+    pysqlite_mutex_enter(sqlite3_db_mutex(db));
+}
+
+static inline void
+pysqlite_db_mutex_leave(sqlite3 *db)
+{
+    pysqlite_mutex_leave(sqlite3_db_mutex(db));
+}
+
+static inline int
+pysqlite_get_autocommit(sqlite3 *db)
+{
+    pysqlite_db_mutex_enter(db);
+    int autocommit = sqlite3_get_autocommit(db);
+    pysqlite_db_mutex_leave(db);
+    return autocommit;
+}
+
+static inline int
+pysqlite_limit(sqlite3 *db, int id, int new_value)
+{
+    pysqlite_db_mutex_enter(db);
+    int old_value = sqlite3_limit(db, id, new_value);
+    pysqlite_db_mutex_leave(db);
+    return old_value;
+}
+
 #endif
diff --git a/PCbuild/_sqlite3.vcxproj b/PCbuild/_sqlite3.vcxproj
index 9ae0a0fc3a009db..8fb964f87efe53c 100644
--- a/PCbuild/_sqlite3.vcxproj
+++ b/PCbuild/_sqlite3.vcxproj
@@ -94,7 +94,7 @@
   <ItemDefinitionGroup>
     <ClCompile>
       <AdditionalIncludeDirectories>$(sqlite3Dir);%(AdditionalIncludeDirectories)</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>PY_SQLITE_HAVE_SERIALIZE;PY_SQLITE_ENABLE_LOAD_EXTENSION;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>PY_SQLITE_HAVE_MUTEX;PY_SQLITE_HAVE_SERIALIZE;PY_SQLITE_ENABLE_LOAD_EXTENSION;%(PreprocessorDefinitions)</PreprocessorDefinitions>
     </ClCompile>
   </ItemDefinitionGroup>
   <ItemGroup>
diff --git a/configure b/configure
index 1b81c6022bfbfdb..b58e10ca13b5bc2 100755
--- a/configure
+++ b/configure
@@ -18279,6 +18279,58 @@ then :
 printf "%s\n" "#define PY_SQLITE_HAVE_SERIALIZE 1" >>confdefs.h
 
 
+fi
+
+      { printf "%s\n" "$as_me:${as_lineno-$LINENO}: checking for sqlite3_mutex_try in -lsqlite3" >&5
+printf %s "checking for sqlite3_mutex_try in -lsqlite3... " >&6; }
+if test ${ac_cv_lib_sqlite3_sqlite3_mutex_try+y}
+then :
+  printf %s "(cached) " >&6
+else case e in #(
+  e) ac_check_lib_save_LIBS=$LIBS
+LIBS="-lsqlite3  $LIBS"
+cat confdefs.h - <<_ACEOF >conftest.$ac_ext
+/* end confdefs.h.  */
+
+/* Override any GCC internal prototype to avoid an error.
+   Use char because int might match the return type of a GCC
+   builtin and then its argument prototype would still apply.
+   The 'extern "C"' is for builds by C++ compilers;
+   although this is not generally supported in C code supporting it here
+   has little cost and some practical benefit (sr 110532).  */
+#ifdef __cplusplus
+extern "C"
+#endif
+char sqlite3_mutex_try (void);
+int
+main (void)
+{
+return sqlite3_mutex_try ();
+  ;
+  return 0;
+}
+_ACEOF
+if ac_fn_c_try_link "$LINENO"
+then :
+  ac_cv_lib_sqlite3_sqlite3_mutex_try=yes
+else case e in #(
+  e) ac_cv_lib_sqlite3_sqlite3_mutex_try=no ;;
+esac
+fi
+rm -f core conftest.err conftest.$ac_objext conftest.beam \
+    conftest$ac_exeext conftest.$ac_ext
+LIBS=$ac_check_lib_save_LIBS ;;
+esac
+fi
+{ printf "%s\n" "$as_me:${as_lineno-$LINENO}: result: $ac_cv_lib_sqlite3_sqlite3_mutex_try" >&5
+printf "%s\n" "$ac_cv_lib_sqlite3_sqlite3_mutex_try" >&6; }
+if test "x$ac_cv_lib_sqlite3_sqlite3_mutex_try" = xyes
+then :
+
+
+printf "%s\n" "#define PY_SQLITE_HAVE_MUTEX 1" >>confdefs.h
+
+
 fi
 
 
diff --git a/configure.ac b/configure.ac
index fa076d486c29607..98c553f372606ac 100644
--- a/configure.ac
+++ b/configure.ac
@@ -4846,6 +4846,12 @@ dnl hence CPPFLAGS instead of CFLAGS.
           [Define if SQLite was compiled with the serialize API]
         )
       ])
+      AC_CHECK_LIB([sqlite3], [sqlite3_mutex_try], [
+        AC_DEFINE(
+          [PY_SQLITE_HAVE_MUTEX], [1],
+          [Define if SQLite was compiled with mutexes (SQLITE_THREADSAFE != 0)]
+        )
+      ])
     ], [
       have_supported_sqlite3=no
     ])
diff --git a/pyconfig.h.in b/pyconfig.h.in
index c575cbf0bf1c779..b9346b00b5822d6 100644
--- a/pyconfig.h.in
+++ b/pyconfig.h.in
@@ -1812,6 +1812,9 @@
 /* Define to 1 to build the sqlite module with loadable extensions support. */
 #undef PY_SQLITE_ENABLE_LOAD_EXTENSION
 
+/* Define if SQLite was compiled with mutexes (SQLITE_THREADSAFE != 0) */
+#undef PY_SQLITE_HAVE_MUTEX
+
 /* Define if SQLite was compiled with the serialize API */
 #undef PY_SQLITE_HAVE_SERIALIZE