correctly fix sqlite3's executemany

Author: picnixz

Lib/test/test_sqlite3/test_dbapi.py

@@ -733,21 +733,27 @@ class CxWrapper:
     def __init__(self, cx):
         self.cx = cx
 
+    def side_effect(self):
+        self.cx.close()
 
-class EvilParamsWithIter(CxWrapper):
 
+class ParamsCxCloseInIterMany(CxWrapper):
     def __iter__(self):
-        self.cx.close()
-        return iter([(1,)])
+        self.side_effect()
+        return iter([(1,), (2,), (3,)])
 
 
-class EvilParamsWithNext(CxWrapper):
+class ParamsCxCloseInNext(CxWrapper):
+    def __init__(self, cx):
+        super().__init__(cx)
+        self.r = iter(range(10))
+
     def __iter__(self):
         return self
 
     def __next__(self):
-        self.cx.close()
-        return (1,)
+        self.side_effect()
+        return (next(self.r),)
 
 
 class CursorTests(unittest.TestCase):
@@ -764,16 +770,6 @@ def tearDown(self):
         self.cu.close()
         self.cx.close()
 
-    def do_test_connection_use_after_close(self, method_name, params_class):
-        # Prevent SIGSEGV with iterable of parameters closing the connection.
-        # Regression test for https://github.com/python/cpython/issues/143198.
-        cx = sqlite.connect(":memory:")
-        self.addCleanup(cx.close)
-        cu = cx.cursor()
-        params = params_class(cx)
-        method = operator.methodcaller(method_name, "SELECT ?", params)
-        self.assertRaises(sqlite.ProgrammingError, method, cu)
-
     def test_execute_no_args(self):
         self.cu.execute("delete from test")
 
@@ -845,10 +841,6 @@ def test_execute_non_iterable(self):
             self.cu.execute("insert into test(id) values (?)", 42)
         self.assertEqual(str(cm.exception), 'parameters are of unsupported type')
 
-    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
-    def test_execute_use_after_close(self, params_class):
-        self.do_test_connection_use_after_close("execute", params_class)
-
     def test_execute_wrong_no_of_args1(self):
         # too many parameters
         with self.assertRaises(sqlite.ProgrammingError):
@@ -1066,9 +1058,17 @@ def test_execute_many_not_iterable(self):
         with self.assertRaises(TypeError):
             self.cu.executemany("insert into test(income) values (?)", 42)
 
-    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
+    @subTests("params_class", (ParamsCxCloseInIterMany, ParamsCxCloseInNext))
     def test_executemany_use_after_close(self, params_class):
-        self.do_test_connection_use_after_close("executemany", params_class)
+        # Prevent SIGSEGV with iterable of parameters closing the connection.
+        # Regression test for https://github.com/python/cpython/issues/143198.
+        cx = sqlite.connect(":memory:")
+        cx.execute("create table tmp(a number)")
+        self.addCleanup(cx.close)
+        cu = cx.cursor()
+        msg = r"Cannot operate on a closed database\."
+        with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+            cu.executemany("insert into tmp(a) values (?)", params_class(cx))
 
     def test_fetch_iter(self):
         # Optional DB-API extension.
@@ -1685,15 +1685,6 @@ def tearDown(self):
         self.cur.close()
         self.con.close()
 
-    def do_test_connection_use_after_close(self, method_name, params_class):
-        # Prevent SIGSEGV with iterable of parameters closing the connection.
-        # Regression test for https://github.com/python/cpython/issues/143198.
-        cx = sqlite.connect(":memory:")
-        self.addCleanup(cx.close)
-        params = params_class(cx)
-        method = operator.methodcaller(method_name, "SELECT ?", params)
-        self.assertRaises(sqlite.ProgrammingError, method, cx)
-
     def test_script_string_sql(self):
         cur = self.cur
         cur.executescript("""
@@ -1760,10 +1751,6 @@ def test_connection_execute(self):
         result = self.con.execute("select 5").fetchone()[0]
         self.assertEqual(result, 5, "Basic test of Connection.execute")
 
-    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
-    def test_connection_execute_use_after_close(self, params_class):
-        self.do_test_connection_use_after_close("execute", params_class)
-
     def test_connection_executemany(self):
         con = self.con
         con.execute("create table test(foo)")
@@ -1772,9 +1759,16 @@ def test_connection_executemany(self):
         self.assertEqual(result[0][0], 3, "Basic test of Connection.executemany")
         self.assertEqual(result[1][0], 4, "Basic test of Connection.executemany")
 
-    @subTests("params_class", (EvilParamsWithIter, EvilParamsWithNext))
+    @subTests("params_class", (ParamsCxCloseInIterMany, ParamsCxCloseInNext))
     def test_connection_executemany_use_after_close(self, params_class):
-        self.do_test_connection_use_after_close("executemany", params_class)
+        # Prevent SIGSEGV with iterable of parameters closing the connection.
+        # Regression test for https://github.com/python/cpython/issues/143198.
+        cx = sqlite.connect(":memory:")
+        cx.execute("create table tmp(a number)")
+        self.addCleanup(cx.close)
+        msg = r"Cannot operate on a closed database\."
+        with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+            cx.executemany("insert into tmp(a) values (?)", params_class(cx))
 
     def test_connection_executescript(self):
         con = self.con

Misc/NEWS.d/next/Library/2025-12-27-10-36-18.gh-issue-143198.DdIHyC.rst

@@ -1,5 +1,3 @@
-:mod:`sqlite3`: fix crashes in :meth:`Connection.execute <sqlite3.Connection.execute>`
-and :meth:`Connection.executemany <sqlite3.Connection.executemany>` when iterating over
-the query's parameters closes the current connection. A similar issue has been fixed
-for the corresponding methods on :class:`~sqlite3.Cursor` objects.
-Patch by Bénédikt Tran.
+:mod:`sqlite3`: fix crashes in :meth:`Connection.executemany <sqlite3.Connection.executemany>`
+and :meth:`Cursor.executemany <sqlite3.Cursor.executemany>` when iterating over
+the query's parameters closes the current connection. Patch by Bénédikt Tran.

Modules/_sqlite/cursor.c

@@ -931,6 +931,11 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
         if (!parameters) {
             break;
         }
+        // PyIter_Next() may have a side-effect on the connection's state.
+        // See: https://github.com/python/cpython/issues/143198.
+        if (!pysqlite_check_connection(self->connection)) {
+            goto error;
+        }
 
         bind_parameters(state, self->statement, parameters);
         if (PyErr_Occurred()) {