HackBase/python-cx_Oracle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add support for specifying an integer for the parameters argument to

Browse Source 
cursor.executemany(). This allows for batch execution when no parameters are
required or when parameters have previously been bound. This replaces
cursor.executemanyprepared() which is now deprecated and will be removed in
cx_Oracle 7.
This commit is contained in:
Anthony Tuininga 2018-05-09 19:35:00 -06:00
parent d9c50574f5
commit 2272af1563
3 changed files with 82 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
doc/src/cursor.rst
View File

@ -182,6 +182,10 @@ Cursor Object
size of each element being allocated in the buffer. If you receive this
error, decrease the number of elements in the sequence parameters.
If there are no parameters, or parameters have previously been bound, the
number of iterations can be specified as an integer instead of needing to
provide a list of empty mappings or sequences.
When true, the batcherrors parameter enables batch error support within
Oracle and ensures that the call succeeds even if an exception takes place
in one or more of the sequence of parameters. The errors can then be
@ -214,6 +218,10 @@ Cursor Object
The DB API definition does not define this method.
.. deprecated:: 6.4
Use :meth:`~Cursor.executemany()` instead with None for the statement
argument and an integer for the parameters argument.
.. method:: Cursor.fetchall()

44
src/cxoCursor.c
View File

@ -1406,7 +1406,8 @@ static PyObject *cxoCursor_execute(cxoCursor *cursor, PyObject *args,
//-----------------------------------------------------------------------------
// cxoCursor_executeMany()
// Execute the statement many times. The number of times is equivalent to the
// number of elements in the array of dictionaries.
// number of elements in the list of parameters, or the provided integer if no
// parameters are required.
//-----------------------------------------------------------------------------
static PyObject *cxoCursor_executeMany(cxoCursor *cursor, PyObject *args,
PyObject *keywordArgs)
@ -1414,15 +1415,22 @@ static PyObject *cxoCursor_executeMany(cxoCursor *cursor, PyObject *args,
static char *keywordList[] = { "statement", "parameters", "batcherrors",
"arraydmlrowcounts", NULL };
int arrayDMLRowCountsEnabled = 0, batchErrorsEnabled = 0;
PyObject *arguments, *listOfArguments, *statement;
PyObject *arguments, *parameters, *statement;
uint32_t mode, i, numRows;
int status;
// expect statement text (optional) plus list of sequences/mappings
if (!PyArg_ParseTupleAndKeywords(args, keywordArgs, "OO!|ii", keywordList,
&statement, &PyList_Type, &listOfArguments,
&batchErrorsEnabled, &arrayDMLRowCountsEnabled))
// validate parameters
if (!PyArg_ParseTupleAndKeywords(args, keywordArgs, "OO|ii", keywordList,
&statement, &parameters, &batchErrorsEnabled,
&arrayDMLRowCountsEnabled))
return NULL;
if (!PyList_Check(parameters) && !PyInt_Check(parameters)) {
PyErr_SetString(PyExc_TypeError,
"parameters should be a list of sequences/dictionaries "
"or an integer specifying the number of times to execute "
"the statement");
return NULL;
}
// make sure the cursor is open
if (cxoCursor_isOpen(cursor) < 0)
@ -1440,16 +1448,20 @@ static PyObject *cxoCursor_executeMany(cxoCursor *cursor, PyObject *args,
if (cxoCursor_internalPrepare(cursor, statement, NULL) < 0)
return NULL;
// perform binds
numRows = (uint32_t) PyList_GET_SIZE(listOfArguments);
for (i = 0; i < numRows; i++) {
arguments = PyList_GET_ITEM(listOfArguments, i);
if (!PyDict_Check(arguments) && !PySequence_Check(arguments))
return cxoError_raiseFromString(cxoInterfaceErrorException,
"expecting a list of dictionaries or sequences");
if (cxoCursor_setBindVariables(cursor, arguments, numRows, i,
(i < numRows - 1)) < 0)
return NULL;
// perform binds, as required
if (PyInt_Check(parameters))
numRows = (uint32_t) PyInt_AsLong(parameters);
else {
numRows = (uint32_t) PyList_GET_SIZE(parameters);
for (i = 0; i < numRows; i++) {
arguments = PyList_GET_ITEM(parameters, i);
if (!PyDict_Check(arguments) && !PySequence_Check(arguments))
return cxoError_raiseFromString(cxoInterfaceErrorException,
"expecting a list of dictionaries or sequences");
if (cxoCursor_setBindVariables(cursor, arguments, numRows, i,
(i < numRows - 1)) < 0)
return NULL;
}
}
if (cxoCursor_performBind(cursor) < 0)
return NULL;

46
test/Cursor.py
View File

@ -184,6 +184,52 @@ class TestCursor(BaseTestCase):
statement, rows)
self.assertEqual(self.cursor.rowcount, 3)
def testExecuteManyWithInvalidParameters(self):
"test calling executemany() with invalid parameters"
self.assertRaises(TypeError, self.cursor.executemany,
"insert into TestTempTable values (:1, :2)",
"These are not valid parameters")
def testExecuteManyNoParameters(self):
"test calling executemany() without any bind parameters"
numRows = 5
self.cursor.execute("truncate table TestTempTable")
self.cursor.executemany("""
declare
t_Id number;
begin
select nvl(count(*), 0) + 1 into t_Id
from TestTempTable;
insert into TestTempTable
values (t_Id, 'Test String ' || t_Id);
end;""", numRows)
self.cursor.execute("select count(*) from TestTempTable")
count, = self.cursor.fetchone()
self.assertEqual(count, numRows)
def testExecuteManyBoundEarlier(self):
"test calling executemany() with binds performed earlier"
numRows = 9
self.cursor.execute("truncate table TestTempTable")
var = self.cursor.var(int, arraysize = numRows)
self.cursor.setinputsizes(var)
self.cursor.executemany("""
declare
t_Id number;
begin
select nvl(count(*), 0) + 1 into t_Id
from TestTempTable;
insert into TestTempTable
values (t_Id, 'Test String ' || t_Id);
select sum(IntCol) into :1
from TestTempTable;
end;""", numRows)
expectedData = [1, 3, 6, 10, 15, 21, 28, 36, 45]
self.assertEqual(var.values, expectedData)
def testPrepare(self):
"""test preparing a statement and executing it multiple times"""
self.assertEqual(self.cursor.statement, None)