python · JelleZijlstra · Mar 17, 2022 · Aug 9, 2021 · Aug 9, 2021 · Aug 9, 2021
@@ -652,8 +652,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")

@@ -319,12 +319,15 @@ 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"]
+        for query in queries:
+            with self.subTest(query=query):
+                self.assertRaisesRegex(sqlite.ProgrammingError, "null char",
+                                       self.con.execute, query)
+            with self.subTest(query=query):
+                self.assertRaisesRegex(sqlite.ProgrammingError, "null char",
+                                       cur.execute, query)

    def test_surrogates(self):
        con = sqlite.connect(":memory:")

@@ -196,6 +196,8 @@ def setUp(self):
        self.con.create_function("returnlonglong", 0, func_returnlonglong)
        self.con.create_function("returnnan", 0, lambda: float("nan"))
        self.con.create_function("returntoolargeint", 0, lambda: 1 << 65)
+        self.con.create_function("return_noncont_blob", 0,
+                                 lambda: memoryview(b"blob")[::2])
        self.con.create_function("raiseexception", 0, func_raiseexception)
        self.con.create_function("memoryerror", 0, func_memoryerror)
        self.con.create_function("overflowerror", 0, func_overflowerror)
@@ -340,10 +342,17 @@ def test_too_large_int(self):
                               "select spam(?)", (1 << 65,))

    def test_non_contiguous_blob(self):
-        self.assertRaisesRegex(ValueError, "could not convert BLOB to buffer",
+        self.assertRaisesRegex(BufferError,
+                               "underlying buffer is not C-contiguous",
                               self.con.execute, "select spam(?)",
                               (memoryview(b"blob")[::2],))

+    @with_tracebacks(BufferError, regex="buffer.*contiguous")
+    def test_return_non_contiguous_blob(self):
+        with self.assertRaises(sqlite.OperationalError):
+            cur = self.con.execute("select return_noncont_blob()")
+            cur.fetchone()
+
    def test_param_surrogates(self):
        self.assertRaisesRegex(UnicodeEncodeError, "surrogates not allowed",
                               self.con.execute, "select spam(?)",
@@ -466,6 +475,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):

diff --git a/Misc/NEWS.d/next/Library/2021-08-10-00-05-53.bpo-44859.9e9_3V.rst b/Misc/NEWS.d/next/Library/2021-08-10-00-05-53.bpo-44859.9e9_3V.rst
@@ -0,0 +1,10 @@
+Raise more accurate and :pep:`249` compatible exceptions in :mod:`sqlite3`.
+
+* Raise :exc:`~sqlite3.InterfaceError` instead of
+  :exc:`~sqlite3.ProgrammingError` for ``SQLITE_MISUSE`` errors.
+* Don't overwrite :exc:`BufferError` with :exc:`ValueError` when conversion to
+  BLOB fails.
+* Raise :exc:`~sqlite3.ProgrammingError` instead of :exc:`~sqlite3.Warning` if
+  user tries to :meth:`~sqlite3.Cursor.execute()` more than one SQL statement.
+* Raise :exc:`~sqlite3.ProgrammingError` instead of :exc:`ValueError` if an SQL
+  query contains null characters.
@@ -578,8 +578,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) {
@@ -591,6 +589,11 @@ _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,
+                     "User-defined functions cannot return '%s' values to "
+                     "SQLite",
+                     Py_TYPE(py_val)->tp_name);
        return -1;
    }
    return 0;

@@ -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->ProgrammingError,
                        "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;
    }
@@ -190,7 +190,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) {