diff --git a/Lib/test/test_sqlite3/test_regression.py b/Lib/test/test_sqlite3/test_regression.py
index 50cced3891d13a5..8a394ca72becdcc 100644
--- a/Lib/test/test_sqlite3/test_regression.py
+++ b/Lib/test/test_sqlite3/test_regression.py
@@ -503,5 +503,72 @@ def test_recursive_cursor_iter(self):
                                    self.cur.fetchall)
 
 
+class CallbackClosesConnectionTests(unittest.TestCase):
+    """Regression tests for gh-151030: callbacks that close the connection
+    during query execution must raise ProgrammingError, not crash."""
+
+    def _make_con(self) -> sqlite.Connection:
+        con = sqlite.connect(":memory:")
+        con.execute("CREATE TABLE t (v INTEGER)")
+        con.execute("INSERT INTO t VALUES (1)")
+        con.commit()
+        return con
+
+    def test_udf_closes_connection(self) -> None:
+        con = self._make_con()
+
+        def bad(x: int) -> int:
+            con.close()
+            return x
+
+        con.create_function("bad", 1, bad)
+        with self.assertRaises((sqlite.ProgrammingError, sqlite.OperationalError)):
+            con.execute("SELECT bad(v) FROM t").fetchall()
+
+    def test_progress_handler_closes_connection(self) -> None:
+        con = self._make_con()
+        fired = False
+
+        def handler() -> int:
+            nonlocal fired
+            if not fired:
+                fired = True
+                con.close()
+            return 0
+
+        con.set_progress_handler(handler, 1)
+        with self.assertRaises((sqlite.ProgrammingError, sqlite.OperationalError)):
+            con.execute("SELECT v FROM t").fetchall()
+
+    def test_trace_callback_closes_connection(self) -> None:
+        con = self._make_con()
+        fired = False
+
+        def tracer(statement: str) -> None:
+            nonlocal fired
+            if not fired:
+                fired = True
+                con.close()
+
+        con.set_trace_callback(tracer)
+        with self.assertRaises((sqlite.ProgrammingError, sqlite.OperationalError)):
+            con.execute("SELECT v FROM t").fetchall()
+
+    def test_authorizer_closes_connection(self) -> None:
+        con = self._make_con()
+        fired = False
+
+        def auth(action: int, arg1: str, arg2: str, db: str, trigger: str) -> int:
+            nonlocal fired
+            if not fired:
+                fired = True
+                con.close()
+            return sqlite.SQLITE_OK
+
+        con.set_authorizer(auth)
+        with self.assertRaises((sqlite.ProgrammingError, sqlite.OperationalError)):
+            con.execute("SELECT v FROM t").fetchall()
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/Misc/NEWS.d/next/Library/2026-06-06-16-39-00.gh-issue-151030.aB3cD4.rst b/Misc/NEWS.d/next/Library/2026-06-06-16-39-00.gh-issue-151030.aB3cD4.rst
new file mode 100644
index 000000000000000..3fe33e2ccece4f5
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2026-06-06-16-39-00.gh-issue-151030.aB3cD4.rst
@@ -0,0 +1,2 @@
+Fix crashes in :mod:`sqlite3` when a callback closes the connection
+while a query is being prepared or executed.
diff --git a/Modules/_sqlite/connection.c b/Modules/_sqlite/connection.c
index 892740b05e55c98..e43a4a887ff4bff 100644
--- a/Modules/_sqlite/connection.c
+++ b/Modules/_sqlite/connection.c
@@ -456,6 +456,10 @@ connection_close(pysqlite_Connection *self)
     sqlite3 *db = self->db;
     self->db = NULL;
 
+    /* Unregister callbacks before closing so that SQLite cannot invoke them
+     * again after free_callback_contexts releases their contexts. */
+    remove_callbacks(db);
+
     Py_BEGIN_ALLOW_THREADS
     /* The v2 close call always returns SQLITE_OK if given a valid database
      * pointer (which we do), so we can safely ignore the return value */
diff --git a/Modules/_sqlite/cursor.c b/Modules/_sqlite/cursor.c
index 5a61e43617984d9..24256c56624b233 100644
--- a/Modules/_sqlite/cursor.c
+++ b/Modules/_sqlite/cursor.c
@@ -360,6 +360,12 @@ _pysqlite_fetch_one_row(pysqlite_Cursor* self)
         return NULL;
 
     sqlite3 *db = self->connection->db;
+    if (db == NULL) {
+        pysqlite_state *state = self->connection->state;
+        PyErr_SetString(state->ProgrammingError,
+                        "Cannot operate on a closed database.");
+        goto error;
+    }
     for (i = 0; i < numcols; i++) {
         if (self->connection->detect_types
                 && self->row_cast_map != NULL
@@ -530,10 +536,16 @@ begin_transaction(pysqlite_Connection *self)
 static PyObject *
 get_statement_from_cache(pysqlite_Cursor *self, PyObject *operation)
 {
-    PyObject *args[] = { NULL, operation, };  // Borrowed ref.
-    PyObject *cache = self->connection->statement_cache;
+    PyObject *args[] = { NULL, operation, };
+    
+    /* Hold a strong reference: a Python callback invoked during statement
+     * preparation (e.g. an authorizer) may close the connection, freeing
+     * the cache while the call is still in progress. */
+    PyObject *cache = Py_NewRef(self->connection->statement_cache);
     size_t nargsf = 1 | PY_VECTORCALL_ARGUMENTS_OFFSET;
-    return PyObject_Vectorcall(cache, args + 1, nargsf, NULL);
+    PyObject *result = PyObject_Vectorcall(cache, args + 1, nargsf, NULL);
+    Py_DECREF(cache);
+    return result;
 }
 
 static inline int
@@ -957,7 +969,7 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
         }
 
         if (rc == SQLITE_DONE) {
-            if (self->statement->is_dml) {
+            if (self->statement->is_dml && self->connection->db) {
                 self->rowcount += (long)sqlite3_changes(self->connection->db);
             }
             if (stmt_reset(self->statement) != SQLITE_OK) {
@@ -967,7 +979,7 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
         Py_XDECREF(parameters);
     }
 
-    if (!multiple) {
+    if (!multiple && self->connection->db) {
         sqlite_int64 lastrowid;
 
         Py_BEGIN_ALLOW_THREADS
@@ -1157,7 +1169,7 @@ pysqlite_cursor_iternext(PyObject *op)
     }
     int rc = stmt_step(stmt);
     if (rc == SQLITE_DONE) {
-        if (self->statement->is_dml) {
+        if (self->statement->is_dml && self->connection->db) {
             self->rowcount = (long)sqlite3_changes(self->connection->db);
         }
         rc = stmt_reset(self->statement);
diff --git a/Modules/_sqlite/util.c b/Modules/_sqlite/util.c
index 177e0f668a0c59e..e990068f8a355e5 100644
--- a/Modules/_sqlite/util.c
+++ b/Modules/_sqlite/util.c
@@ -138,6 +138,11 @@ set_error_from_code(pysqlite_state *state, int code)
 int
 set_error_from_db(pysqlite_state *state, sqlite3 *db)
 {
+    if (db == NULL) {
+        PyErr_SetString(state->ProgrammingError,
+                        "Cannot operate on a closed database.");
+        return SQLITE_MISUSE;
+    }
     int errorcode = sqlite3_errcode(db);
     PyObject *exc_class = get_exception_class(state, errorcode);
     if (exc_class == NULL) {