mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-03 00:23:06 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

[3.9] bpo-43853: Handle sqlite3_value_text() errors (GH-25422). (GH-27627)

Browse source 
(cherry picked from commit 006fd869e4)
This commit is contained in:
Erlend Egeberg Aasland 2021-08-06 19:57:39 +02:00 • committed by GitHub
parent 8c17db6cd4
commit 8c07fef867
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 26 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

12
Lib/sqlite3/test/userfunctions.py
View file

@ -236,9 +236,11 @@ class FunctionTests(unittest.TestCase):
def CheckParamString(self): def CheckParamString(self):
cur = self.con.cursor() cur = self.con.cursor()
cur.execute("select isstring(?)", ("foo",)) for text in ["foo", str()]:
val = cur.fetchone()[0] with self.subTest(text=text):
self.assertEqual(val, 1) cur.execute("select isstring(?)", (text,))
val = cur.fetchone()[0]
self.assertEqual(val, 1)
def CheckParamInt(self): def CheckParamInt(self):
cur = self.con.cursor() cur = self.con.cursor()
@ -387,9 +389,9 @@ class AggregateTests(unittest.TestCase):
def CheckAggrCheckParamStr(self): def CheckAggrCheckParamStr(self):
cur = self.con.cursor() cur = self.con.cursor()
cur.execute("select checkType('str', ?)", ("foo",)) cur.execute("select checkTypes('str', ?, ?)", ("foo", str()))
val = cur.fetchone()[0] val = cur.fetchone()[0]
self.assertEqual(val, 1) self.assertEqual(val, 2)
def CheckAggrCheckParamInt(self): def CheckAggrCheckParamInt(self):
cur = self.con.cursor() cur = self.con.cursor()

3
Misc/NEWS.d/next/Library/2021-04-15-12-02-17.bpo-43853.XXCVAp.rst Normal file
View file

@ -0,0 +1,3 @@
Improve :mod:`sqlite3` error handling: ``sqlite3_value_text()`` errors that
set ``SQLITE_NOMEM`` now raise :exc:`MemoryError`. Patch by Erlend E.
Aasland.

27
Modules/_sqlite/connection.c
View file

@ -546,7 +546,6 @@ PyObject* _pysqlite_build_py_params(sqlite3_context *context, int argc, sqlite3_
int i; int i;
sqlite3_value* cur_value; sqlite3_value* cur_value;
PyObject* cur_py_value; PyObject* cur_py_value;
const char* val_str;
Py_ssize_t buflen; Py_ssize_t buflen;
args = PyTuple_New(argc); args = PyTuple_New(argc);
@ -563,16 +562,19 @@ PyObject* _pysqlite_build_py_params(sqlite3_context *context, int argc, sqlite3_
case SQLITE_FLOAT: case SQLITE_FLOAT:
cur_py_value = PyFloat_FromDouble(sqlite3_value_double(cur_value)); cur_py_value = PyFloat_FromDouble(sqlite3_value_double(cur_value));
break; break;
case SQLITE_TEXT: case SQLITE_TEXT: {
val_str = (const char*)sqlite3_value_text(cur_value); sqlite3 *db = sqlite3_context_db_handle(context);
cur_py_value = PyUnicode_FromString(val_str); const char *text = (const char *)sqlite3_value_text(cur_value);
/* TODO: have a way to show errors here */
if (!cur_py_value) { if (text == NULL && sqlite3_errcode(db) == SQLITE_NOMEM) {
PyErr_Clear(); PyErr_NoMemory();
Py_INCREF(Py_None); goto error;
cur_py_value = Py_None;
} }
Py_ssize_t size = sqlite3_value_bytes(cur_value);
cur_py_value = PyUnicode_FromStringAndSize(text, size);
break; break;
}
case SQLITE_BLOB: case SQLITE_BLOB:
buflen = sqlite3_value_bytes(cur_value); buflen = sqlite3_value_bytes(cur_value);
cur_py_value = PyBytes_FromStringAndSize( cur_py_value = PyBytes_FromStringAndSize(
@ -585,8 +587,7 @@ PyObject* _pysqlite_build_py_params(sqlite3_context *context, int argc, sqlite3_
} }
if (!cur_py_value) { if (!cur_py_value) {
Py_DECREF(args); goto error;
return NULL;
} }
PyTuple_SetItem(args, i, cur_py_value); PyTuple_SetItem(args, i, cur_py_value);
@ -594,6 +595,10 @@ PyObject* _pysqlite_build_py_params(sqlite3_context *context, int argc, sqlite3_
} }
return args; return args;
error:
Py_DECREF(args);
return NULL;
} }
void _pysqlite_func_callback(sqlite3_context* context, int argc, sqlite3_value** argv) void _pysqlite_func_callback(sqlite3_context* context, int argc, sqlite3_value** argv)