If the *size* parameter is used, then it is best for it to retain the same
value from one :meth:`fetchmany` call to the next.
+ .. versionchanged:: next
+ Negative *size* values are rejected by raising :exc:`ValueError`.
+
.. method:: fetchall()
Return all (remaining) rows of a query result as a :class:`list`.
Read/write attribute that controls the number of rows returned by :meth:`fetchmany`.
The default value is 1 which means a single row would be fetched per call.
+ .. versionchanged:: next
+ Negative values are rejected by raising :exc:`ValueError`.
+
.. attribute:: connection
Read-only attribute that provides the SQLite database :class:`Connection`
# 3. This notice may not be removed or altered from any source distribution.
import contextlib
+import functools
import os
import sqlite3 as sqlite
import subprocess
# now set to 2
self.cu.arraysize = 2
- # now make the query return 3 rows
+ # now make the query return 2 rows from a table of 3 rows
self.cu.execute("delete from test")
self.cu.execute("insert into test(name) values ('A')")
self.cu.execute("insert into test(name) values ('B')")
self.assertEqual(len(res), 2)
+ def test_invalid_array_size(self):
+ UINT32_MAX = (1 << 32) - 1
+ setter = functools.partial(setattr, self.cu, 'arraysize')
+
+ self.assertRaises(TypeError, setter, 1.0)
+ self.assertRaises(ValueError, setter, -3)
+ self.assertRaises(OverflowError, setter, UINT32_MAX + 1)
+
def test_fetchmany(self):
+ # no active SQL statement
+ res = self.cu.fetchmany()
+ self.assertEqual(res, [])
+ res = self.cu.fetchmany(1000)
+ self.assertEqual(res, [])
+
+ # test default parameter
+ self.cu.execute("select name from test")
+ res = self.cu.fetchmany()
+ self.assertEqual(len(res), 1)
+
+ # test when the number of requested rows exceeds the actual count
self.cu.execute("select name from test")
res = self.cu.fetchmany(100)
self.assertEqual(len(res), 1)
res = self.cu.fetchmany(100)
self.assertEqual(res, [])
+ # test when size = 0
+ self.cu.execute("select name from test")
+ res = self.cu.fetchmany(0)
+ self.assertEqual(res, [])
+ res = self.cu.fetchmany(100)
+ self.assertEqual(len(res), 1)
+ res = self.cu.fetchmany(100)
+ self.assertEqual(res, [])
+
+ def test_invalid_fetchmany(self):
+ UINT32_MAX = (1 << 32) - 1
+ fetchmany = self.cu.fetchmany
+
+ self.assertRaises(TypeError, fetchmany, 1.0)
+ self.assertRaises(ValueError, fetchmany, -3)
+ self.assertRaises(OverflowError, fetchmany, UINT32_MAX + 1)
+
def test_fetchmany_kw_arg(self):
"""Checks if fetchmany works with keyword arguments"""
self.cu.execute("select name from test")
--- /dev/null
+:mod:`sqlite3`: correctly handle maximum number of rows to fetch in
+:meth:`Cursor.fetchmany <sqlite3.Cursor.fetchmany>` and reject negative
+values for :attr:`Cursor.arraysize <sqlite3.Cursor.arraysize>`. Patch by
+Bénédikt Tran.
# include "pycore_gc.h" // PyGC_Head
# include "pycore_runtime.h" // _Py_ID()
#endif
+#include "pycore_long.h" // _PyLong_Size_t_Converter()
#include "pycore_modsupport.h" // _PyArg_CheckPositional()
static int
{"fetchmany", _PyCFunction_CAST(pysqlite_cursor_fetchmany), METH_FASTCALL|METH_KEYWORDS, pysqlite_cursor_fetchmany__doc__},
static PyObject *
-pysqlite_cursor_fetchmany_impl(pysqlite_Cursor *self, int maxrows);
+pysqlite_cursor_fetchmany_impl(pysqlite_Cursor *self, size_t maxrows);
static PyObject *
pysqlite_cursor_fetchmany(pysqlite_Cursor *self, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
#undef KWTUPLE
PyObject *argsbuf[1];
Py_ssize_t noptargs = nargs + (kwnames ? PyTuple_GET_SIZE(kwnames) : 0) - 0;
- int maxrows = self->arraysize;
+ size_t maxrows = ((pysqlite_Cursor *)self)->arraysize;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser, 0, 1, 0, argsbuf);
if (!args) {
if (!noptargs) {
goto skip_optional_pos;
}
- maxrows = PyLong_AsInt(args[0]);
- if (maxrows == -1 && PyErr_Occurred()) {
+ if (!_PyLong_Size_t_Converter(args[0], &maxrows)) {
goto exit;
}
skip_optional_pos:
{
return pysqlite_cursor_close_impl(self);
}
-/*[clinic end generated code: output=a8ce095c3c80cf65 input=a9049054013a1b77]*/
+
+#if !defined(_sqlite3_Cursor_arraysize_DOCSTR)
+# define _sqlite3_Cursor_arraysize_DOCSTR NULL
+#endif
+#if defined(_SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF)
+# undef _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF
+# define _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF {"arraysize", (getter)_sqlite3_Cursor_arraysize_get, (setter)_sqlite3_Cursor_arraysize_set, _sqlite3_Cursor_arraysize_DOCSTR},
+#else
+# define _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF {"arraysize", (getter)_sqlite3_Cursor_arraysize_get, NULL, _sqlite3_Cursor_arraysize_DOCSTR},
+#endif
+
+static PyObject *
+_sqlite3_Cursor_arraysize_get_impl(pysqlite_Cursor *self);
+
+static PyObject *
+_sqlite3_Cursor_arraysize_get(pysqlite_Cursor *self, void *Py_UNUSED(context))
+{
+ return _sqlite3_Cursor_arraysize_get_impl(self);
+}
+
+#if !defined(_sqlite3_Cursor_arraysize_DOCSTR)
+# define _sqlite3_Cursor_arraysize_DOCSTR NULL
+#endif
+#if defined(_SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF)
+# undef _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF
+# define _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF {"arraysize", (getter)_sqlite3_Cursor_arraysize_get, (setter)_sqlite3_Cursor_arraysize_set, _sqlite3_Cursor_arraysize_DOCSTR},
+#else
+# define _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF {"arraysize", NULL, (setter)_sqlite3_Cursor_arraysize_set, NULL},
+#endif
+
+static int
+_sqlite3_Cursor_arraysize_set_impl(pysqlite_Cursor *self, PyObject *value);
+
+static int
+_sqlite3_Cursor_arraysize_set(pysqlite_Cursor *self, PyObject *value, void *Py_UNUSED(context))
+{
+ int return_value;
+
+ return_value = _sqlite3_Cursor_arraysize_set_impl(self, value);
+
+ return return_value;
+}
+/*[clinic end generated code: output=332c298249ae57b0 input=a9049054013a1b77]*/
/*[clinic input]
_sqlite3.Cursor.fetchmany as pysqlite_cursor_fetchmany
- size as maxrows: int(c_default='self->arraysize') = 1
+ size as maxrows: size_t(c_default='((pysqlite_Cursor *)self)->arraysize') = 1
The default value is set by the Cursor.arraysize attribute.
Fetches several rows from the resultset.
[clinic start generated code]*/
static PyObject *
-pysqlite_cursor_fetchmany_impl(pysqlite_Cursor *self, int maxrows)
-/*[clinic end generated code: output=a8ef31fea64d0906 input=c26e6ca3f34debd0]*/
+pysqlite_cursor_fetchmany_impl(pysqlite_Cursor *self, size_t maxrows)
+/*[clinic end generated code: output=8bff46d4f24c9e84 input=5485f1f2ae46c2ff]*/
{
+ if (maxrows > (size_t)UINT32_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "Python int too large for C uint32_t");
+ return NULL;
+ }
+
PyObject* row;
PyObject* list;
- int counter = 0;
list = PyList_New(0);
if (!list) {
return NULL;
}
- while ((row = pysqlite_cursor_iternext(self))) {
- if (PyList_Append(list, row) < 0) {
- Py_DECREF(row);
- break;
- }
+ while (maxrows > 0 && (row = pysqlite_cursor_iternext(self))) {
+ int rc = PyList_Append(list, row);
Py_DECREF(row);
-
- if (++counter == maxrows) {
+ if (rc < 0) {
break;
}
+ maxrows--;
}
if (PyErr_Occurred()) {
Py_RETURN_NONE;
}
+/*[clinic input]
+@getter
+_sqlite3.Cursor.arraysize
+[clinic start generated code]*/
+
+static PyObject *
+_sqlite3_Cursor_arraysize_get_impl(pysqlite_Cursor *self)
+/*[clinic end generated code: output=e0919d97175e6c50 input=3278f8d3ecbd90e3]*/
+{
+ return PyLong_FromSize_t(self->arraysize);
+}
+
+/*[clinic input]
+@setter
+_sqlite3.Cursor.arraysize
+[clinic start generated code]*/
+
+static int
+_sqlite3_Cursor_arraysize_set_impl(pysqlite_Cursor *self, PyObject *value)
+/*[clinic end generated code: output=af59a6b09f8cce6e input=ace48cb114e26060]*/
+{
+ size_t converted;
+ if (!_PyLong_Size_t_Converter(value, &converted)) {
+ return -1;
+ }
+ if (converted > (size_t)UINT32_MAX) {
+ PyErr_SetString(PyExc_OverflowError,
+ "Python int too large to convert to C uint32_t");
+ return -1;
+ }
+ self->arraysize = (uint32_t)converted;
+ return 0;
+}
+
static PyMethodDef cursor_methods[] = {
PYSQLITE_CURSOR_CLOSE_METHODDEF
PYSQLITE_CURSOR_EXECUTEMANY_METHODDEF
{
{"connection", _Py_T_OBJECT, offsetof(pysqlite_Cursor, connection), Py_READONLY},
{"description", _Py_T_OBJECT, offsetof(pysqlite_Cursor, description), Py_READONLY},
- {"arraysize", Py_T_INT, offsetof(pysqlite_Cursor, arraysize), 0},
{"lastrowid", _Py_T_OBJECT, offsetof(pysqlite_Cursor, lastrowid), Py_READONLY},
{"rowcount", Py_T_LONG, offsetof(pysqlite_Cursor, rowcount), Py_READONLY},
{"row_factory", _Py_T_OBJECT, offsetof(pysqlite_Cursor, row_factory), 0},
{NULL}
};
+static struct PyGetSetDef cursor_getsets[] = {
+ _SQLITE3_CURSOR_ARRAYSIZE_GETSETDEF
+ {NULL},
+};
+
static const char cursor_doc[] =
PyDoc_STR("SQLite database cursor class.");
{Py_tp_iternext, pysqlite_cursor_iternext},
{Py_tp_methods, cursor_methods},
{Py_tp_members, cursor_members},
+ {Py_tp_getset, cursor_getsets},
{Py_tp_init, pysqlite_cursor_init},
{Py_tp_traverse, cursor_traverse},
{Py_tp_clear, cursor_clear},
pysqlite_Connection* connection;
PyObject* description;
PyObject* row_cast_map;
- int arraysize;
+ uint32_t arraysize;
PyObject* lastrowid;
long rowcount;
PyObject* row_factory;
projects / thirdparty / Python / cpython.git / commitdiff
