bpo-44859: Raise more accurate exceptions in sqlite3

Lib/test/test_sqlite3/test_dbapi.py

@@ -579,8 +579,9 @@ def test_execute_illegal_sql(self):
             self.cu.execute("select asdf")
 
     def test_execute_too_much_sql(self):
-        with self.assertRaises(sqlite.Warning):
-            self.cu.execute("select 5+4; select 4+5")
+        self.assertRaisesRegex(sqlite.ProgrammingError,
+                               "You can only execute one statement at a time",
+                               self.cu.execute, "select 5+4; select 4+5")
 
     def test_execute_too_much_sql2(self):
         self.cu.execute("select 5+4; -- foo bar")

Lib/test/test_sqlite3/test_regression.py

@@ -342,12 +342,16 @@ def test_invalid_isolation_level_type(self):
 
     def test_null_character(self):
         # Issue #21147
-        con = sqlite.connect(":memory:")
-        self.assertRaises(ValueError, con, "\0select 1")
-        self.assertRaises(ValueError, con, "select 1\0")
-        cur = con.cursor()
-        self.assertRaises(ValueError, cur.execute, " \0select 2")
-        self.assertRaises(ValueError, cur.execute, "select 2\0")
+        cur = self.con.cursor()
+        queries = ["\0select 1", "select 1\0"]
+        msg = "the query contains a null character"
+        for query in queries:
+            with self.subTest(query=query):
+                self.assertRaisesRegex(sqlite.DataError, msg,
+                                       self.con.execute, query)
+            with self.subTest(query=query):
+                self.assertRaisesRegex(sqlite.DataError, msg,
+                                       cur.execute, query)
 
     def test_surrogates(self):
         con = sqlite.connect(":memory:")

Lib/test/test_sqlite3/test_userfunctions.py

@@ -461,6 +461,12 @@ def test_func_return_too_large_blob(self, size):
             with self.assertRaises(sqlite.DataError):
                 cur.execute("select largeblob()")
 
+    def test_func_return_illegal_value(self):
+        self.con.create_function("badreturn", 0, lambda: self)
+        msg = "user-defined function raised exception"
+        self.assertRaisesRegex(sqlite.OperationalError, msg,
+                               self.con.execute, "select badreturn()")
+
 
 class AggregateTests(unittest.TestCase):
     def setUp(self):

Misc/NEWS.d/next/Library/2021-08-10-00-05-53.bpo-44859.9e9_3V.rst

@@ -0,0 +1,7 @@
+Raise more accurate exceptions in :mod:`sqlite3`.
+
+* Raise InterfaceError iso. ProgrammingError for SQLITE_MISUSE
+* Don't overwrite BufferError with ValueError when converting to BLOB
+* Raise ProgrammingError iso. Warning if user tries to execute() more than one
+  statement
+* Raise DataError iso. ValueError if query contains NULL characters

Modules/_sqlite/connection.c

@@ -556,8 +556,6 @@ _pysqlite_set_result(sqlite3_context* context, PyObject* py_val)
     } else if (PyObject_CheckBuffer(py_val)) {
         Py_buffer view;
         if (PyObject_GetBuffer(py_val, &view, PyBUF_SIMPLE) != 0) {
-            PyErr_SetString(PyExc_ValueError,
-                            "could not convert BLOB to buffer");
             return -1;
         }
         if (view.len > INT_MAX) {
@@ -569,6 +567,10 @@ _pysqlite_set_result(sqlite3_context* context, PyObject* py_val)
         sqlite3_result_blob(context, view.buf, (int)view.len, SQLITE_TRANSIENT);
         PyBuffer_Release(&view);
     } else {
+        callback_context *ctx = (callback_context *)sqlite3_user_data(context);
+        PyErr_Format(ctx->state->ProgrammingError,
+                     "UDF's cannot return '%s' values to SQLite",
+                     Py_TYPE(py_val)->tp_name);
         return -1;
     }
     return 0;

Modules/_sqlite/statement.c

@@ -67,7 +67,7 @@ pysqlite_statement_create(pysqlite_Connection *connection, PyObject *sql)
         return NULL;
     }
     if (strlen(sql_cstr) != (size_t)size) {
-        PyErr_SetString(PyExc_ValueError,
+        PyErr_SetString(connection->DataError,
                         "the query contains a null character");
         return NULL;
     }
@@ -85,7 +85,7 @@ pysqlite_statement_create(pysqlite_Connection *connection, PyObject *sql)
     }
 
     if (pysqlite_check_remaining_sql(tail)) {
-        PyErr_SetString(connection->Warning,
+        PyErr_SetString(connection->ProgrammingError,
                         "You can only execute one statement at a time.");
         goto error;
     }
@@ -183,7 +183,6 @@ int pysqlite_statement_bind_parameter(pysqlite_Statement* self, int pos, PyObjec
         case TYPE_BUFFER: {
             Py_buffer view;
             if (PyObject_GetBuffer(parameter, &view, PyBUF_SIMPLE) != 0) {
-                PyErr_SetString(PyExc_ValueError, "could not convert BLOB to buffer");
                 return -1;
             }
             if (view.len > INT_MAX) {