Added support for creating a temporary LOB directly.

doc/src/connection.rst

@@ -129,6 +129,19 @@ Connection Object
     Commit any pending transactions to the database.
 
 
+.. method:: Connection.createlob(lobType)
+
+    Create and return a new temporary :ref:`LOB object <lobobj>` of the
+    specified type. The lobType parameter should be one of
+    :data:`cx_Oracle.CLOB`, :data:`cx_Oracle.BLOB` or :data:`cx_Oracle.NCLOB`.
+
+    .. versionadded:: 6.2
+
+    .. note::
+
+        This method is an extension to the DB API definition.
+
+
 .. attribute:: Connection.current_schema
 
     This read-write attribute sets the current schema attribute for the

src/cxoConnection.c

@@ -36,6 +36,7 @@ static PyObject *cxoConnection_contextManagerEnter(cxoConnection*, PyObject*);
 static PyObject *cxoConnection_contextManagerExit(cxoConnection*, PyObject*);
 static PyObject *cxoConnection_changePassword(cxoConnection*, PyObject*);
 static PyObject *cxoConnection_getType(cxoConnection*, PyObject*);
+static PyObject *cxoConnection_createLob(cxoConnection*, PyObject*);
 static PyObject *cxoConnection_getStmtCacheSize(cxoConnection*, void*);
 static PyObject *cxoConnection_newEnqueueOptions(cxoConnection*, PyObject*);
 static PyObject *cxoConnection_newDequeueOptions(cxoConnection*, PyObject*);
@@ -100,6 +101,7 @@ static PyMethodDef cxoConnectionMethods[] = {
             METH_VARARGS | METH_KEYWORDS },
     { "enq", (PyCFunction) cxoConnection_enqueue,
             METH_VARARGS | METH_KEYWORDS },
+    { "createlob", (PyCFunction) cxoConnection_createLob, METH_O },
     { NULL }
 };
 
@@ -901,6 +903,45 @@ static PyObject *cxoConnection_getType(cxoConnection *conn, PyObject *nameObj)
 }
 
 
+//-----------------------------------------------------------------------------
+// cxoConnection_createLob()
+//   Create a new temporary LOB and return it.
+//-----------------------------------------------------------------------------
+static PyObject *cxoConnection_createLob(cxoConnection *conn,
+        PyObject *lobType)
+{
+    dpiOracleTypeNum oracleTypeNum;
+    dpiLob *handle;
+    PyObject *lob;
+
+    // verify connection is open
+    if (cxoConnection_isConnected(conn) < 0)
+        return NULL;
+
+    // verify the LOB type
+    if (lobType == (PyObject*) &cxoPyTypeClobVar)
+        oracleTypeNum = DPI_ORACLE_TYPE_CLOB;
+    else if (lobType == (PyObject*) &cxoPyTypeBlobVar)
+        oracleTypeNum = DPI_ORACLE_TYPE_BLOB;
+    else if (lobType == (PyObject*) &cxoPyTypeNclobVar)
+        oracleTypeNum = DPI_ORACLE_TYPE_NCLOB;
+    else {
+        PyErr_SetString(PyExc_TypeError,
+                "parameter should be one of cx_Oracle.CLOB, cx_Oracle.BLOB "
+                "or cx_Oracle.NCLOB");
+        return NULL;
+    }
+
+    // create a temporary LOB
+    if (dpiConn_newTempLob(conn->handle, oracleTypeNum, &handle) < 0)
+        return cxoError_raiseAndReturnNull();
+    lob = cxoLob_new(conn, oracleTypeNum, handle);
+    if (!lob)
+        dpiLob_release(handle);
+    return lob;
+}
+
+
 //-----------------------------------------------------------------------------
 // cxoConnection_getVersion()
 //   Retrieve the version of the database and return it. Note that this

test/LobVar.py

@@ -95,6 +95,23 @@ class TestLobVar(BaseTestCase):
         lob.trim()
         self.assertEqual(lob.size(), 0)
 
+    def __TestTemporaryLOB(self, lobType):
+        self.cursor.execute("truncate table Test%ss" % lobType)
+        value = "A test string value"
+        if lobType == "BLOB" and sys.version_info[0] >= 3:
+            value = value.encode("ascii")
+        lobTypeObj = getattr(cx_Oracle, lobType)
+        lob = self.connection.createlob(lobTypeObj)
+        lob.write(value)
+        self.cursor.execute("""
+                insert into Test%ss (IntCol, %sCol)
+                values (:intVal, :lobVal)""" % (lobType, lobType),
+                intVal = 1,
+                lobVal = lob)
+        self.cursor.execute("select %sCol from Test%ss" % (lobType, lobType))
+        lob, = self.cursor.fetchone()
+        self.assertEqual(lob.read(), value)
+
     def __ValidateQuery(self, rows, lobType):
         longString = ""
         for row in rows:
@@ -172,6 +189,18 @@ class TestLobVar(BaseTestCase):
         "test operations on CLOBs"
         self.__TestLobOperations("CLOB")
 
+    def testCreateBlob(self):
+        "test creating a temporary BLOB"
+        self.__TestTemporaryLOB("BLOB")
+
+    def testCreateClob(self):
+        "test creating a temporary CLOB"
+        self.__TestTemporaryLOB("CLOB")
+
+    def testCreateNclob(self):
+        "test creating a temporary NCLOB"
+        self.__TestTemporaryLOB("NCLOB")
+
     def testMultipleFetch(self):
         "test retrieving data from a CLOB after multiple fetches"
         self.cursor.arraysize = 1