Examples of output handlers are shown in :ref:`numberprecision`,

:ref:`directlobs` and :ref:`fetching-raw-data`. Also see samples such as `samples/TypeHandlers.py

.. _fetching-raw-data:

Fetching Raw Data

Sometimes cx_Oracle may have problems converting data to unicode and you may

want to inspect the problem closer rather than auto-fix it using the

encodingerrors parameter. This may be useful when a database contains

records or fields that are in a wrong encoding altogether.

It is not recommended to use mixed encodings in databases.

This functionality is aimed at troubleshooting databases

that have inconsistent encodings for external reasons.

For these cases, you can pass in the in additional keyword argument

``bypassencoding = True`` into :meth:`Cursor.var()`. This needs

to be used in combination with :ref:`outputtypehandlers`

.. code-block:: python

#defining output type handlers method

def ConvertStringToBytes(cursor, name, defaultType, size, precision, scale):

if defaultType == cx_Oracle.STRING:

return cursor.var(str, arraysize=cursor.arraysize, bypassencoding = True)

#set cursor outputtypehandler to the method above

cursor = connection.cursor()

ursor.outputtypehandler = ConvertStringToBytes

This will allow you to receive data as raw bytes.

.. code-block:: python

statement = cursor.execute("select content, charset from SomeTable")

data = statement.fetchall()

This will produce output as:

.. code-block:: python

[(b'Fianc\xc3\xa9', b'UTF-8')]

Note that last \xc3\xa9 is é in UTF-8. Then in you can do following:

.. code-block:: python

import codecs

# data = [(b'Fianc\xc3\xa9', b'UTF-8')]

unicodecontent = data[0][0].decode(data[0][1].decode()) # Assuming your charset encoding is UTF-8

This will revert it back to "Fiancé".

If you want to save ``b'Fianc\xc3\xa9'`` to database you will need to create

:meth:`Cursor.var()` that will tell cx_Oracle that the value is indeed

intended as a string:

.. code-block:: python

connection = cx_Oracle.connect("hr", userpwd, "dbhost.example.com/orclpdb1")

cursor = connection.cursor()

cursorvariable = cursor.var(cx_Oracle.STRING)

cursorvariable.setvalue(0, "Fiancé".encode("UTF-8")) # b'Fianc\xc4\x9b'

cursor.execute("update SomeTable set SomeColumn = :param where id = 1", param=cursorvariable)

At that point, the bytes will be assumed to be in the correct encoding and should insert as you expect.

.. warning::

This functionality is "as-is": when saving strings like this,

the bytes will be assumed to be in the correct encoding and will

insert like that. Proper encoding is the responsibility of the user and

no correctness of any data in the database can be assumed

to exist by itself.

import cx_Oracle

import sample_env

def ConvertStringToBytes(cursor, name, defaultType, size, precision, scale):

if defaultType == cx_Oracle.STRING:

return cursor.var(str, arraysize=cursor.arraysize, bypassencoding = True)

connection = cx_Oracle.connect(sample_env.get_main_connect_string())

cursor = connection.cursor()

cursor.outputtypehandler = ConvertStringToBytes

sql = 'create table EncodingExperiment (content varchar2(100), encoding varchar2(15))'

print('Creating experiment table')

try:

cursor.execute(sql)

print('Success, will attempt to add records')

except Exception as err:

# table already exists

print('%s\n%s'%(err, 'EncodingExperiment table exists... Will attempt to add records'))

unicode_string = 'I bought a cafetière on the Champs-Élysées'

windows_1252_encoded = unicode_string.encode('windows-1252')

utf8_encoded = unicode_string.encode('utf-8')

sqlparameters = [(windows_1252_encoded, 'windows-1252'), (utf8_encoded, 'utf-8')]

sql = 'insert into EncodingExperiment (content, encoding) values (:content, :encoding)'

content_variable = cursor.var(cx_Oracle.STRING)

print('Adding records to the table: "EncodingExperiment"')

for sqlparameter in sqlparameters:

content, encoding = sqlparameter

# setting content_variable value to a byte value and instert it as such

content_variable.setvalue(0, content)

cursor.execute(sql, content=content_variable, encoding=encoding)

sql = 'select * from EncodingExperiment'

print('Fetching records from table EncodingExperiment')

result = cursor.execute(sql).fetchall()

for dataset in result:

content, encoding = dataset[0], dataset[1].decode()

decodedcontent = content.decode(encoding)

print('Is "%s" == "%s" ?\nResult: %s, (decoded from: %s)'%(decodedcontent, unicode_string, decodedcontent == unicode_string, encoding))

print('Finished testing, will attempt to drop the table "EncodingExperiment"')

sql = 'drop table EncodingExperiment'

try:

cursor.execute(sql)

print('Successfully droped table "EncodingExperiment" from database.')

except Exception as err:

print('Failed to drop table from the database, info: %s'%err)

"inconverter", "outconverter", "typename", "encodingErrors", "bypassencoding",

int size, arraySize, bypassEncoding;

size = bypassEncoding = 0;

if (!PyArg_ParseTupleAndKeywords(args, keywordArgs, "O|iiOOOz#p",

&typeNameObj, &encodingErrors, &encodingErrorsLength, &bypassEncoding))

// Flag that manually changes transform type to bytes

if (bypassEncoding) {

var->transformNum = CXO_TRANSFORM_BINARY;

}