cx_Oracle: How can I receive each row as a dictionary?

Question

By default, cx_Oracle returns each row as a tuple.

>>> import cx_Oracle
>>> conn=cx_Oracle.connect('scott/tiger')
>>> curs=conn.cursor()
>>> curs.execute("select * from foo");
>>> curs.fetchone()
(33, 'blue')

How can I return each row as a dictionary?

Inextricable

Answer 1

You can override the cursor's rowfactory method. You will need to do this each time you perform the query.

Here's the results of the standard query, a tuple.

curs.execute('select * from foo')
curs.fetchone()
    (33, 'blue')

Returning a named tuple:

def makeNamedTupleFactory(cursor):
    columnNames = [d[0].lower() for d in cursor.description]
    import collections
    Row = collections.namedtuple('Row', columnNames)
    return Row

curs.rowfactory = makeNamedTupleFactory(curs)
curs.fetchone()
    Row(x=33, y='blue')

Returning a dictionary:

def makeDictFactory(cursor):
    columnNames = [d[0] for d in cursor.description]
    def createRow(*args):
        return dict(zip(columnNames, args))
    return createRow

curs.rowfactory = makeDictFactory(curs)
curs.fetchone()
    {'Y': 'brown', 'X': 1}

Credit to Amaury Forgeot d'Arc: http://sourceforge.net/p/cx-oracle/mailman/message/27145597

Inextricable

Answer 2

A very short version:

curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))

Tested on Python 3.7.0 & cx_Oracle 7.1.2

Vertex

Placing this line after the "curs.execute" has worked for me. Returns list of dicts. Thanks. – Ariosto

Can you make this answer complete and independent from the others? – Wert

Answer 3

^{Old question but adding some helpful links with a Python recipe}

According to cx_Oracle documentation:

Cursor.rowfactory

This read-write attribute specifies a method to call for each row that is retrieved from the database. Ordinarily a tuple is returned for each row but if this attribute is set, the method is called with the tuple that would normally be returned, and the result of the method is returned instead.

The cx_Oracle - Python Interface for Oracle Database Also points to GitHub repository for lots of helpful sample examples. Please check GenericRowFactory.py.

Googled: This PPT can be further helpful: ^[PDF]CON6543 Python and Oracle Database - RainFocus

Recipe

Django database backend for Oracle under the hood uses cx_Oracle. In earlier versions ( Django 1.11- ) they have written _rowfactory(cursor, row) That also cast cx_Oracle's numeric data types into relevant Python data and strings into unicode.

If you have installed Django Please check base.py as follows:

$ DJANGO_DIR="$(python -c 'import django, os; print(os.path.dirname(django.__file__))')"
$ vim $DJANGO_DIR/db/backends/oracle/base.py

One can borrow _rowfactory() from $DJANGO_DIR/db/backends/oracle/base.py and can apply below decorator naming to make it return namedtuple instead of simple tuple.

mybase.py

import functools
from itertools import izip, imap
from operator import itemgetter
from collections import namedtuple
import cx_Oracle as Database
import decimal

def naming(rename=False, case=None):
    def decorator(rowfactory):
        @functools.wraps(rowfactory)
        def decorated_rowfactory(cursor, row, typename="GenericRow"):
            field_names = imap(case, imap(itemgetter(0), cursor.description))
            return namedtuple(typename, field_names)._make(rowfactory(cursor, row))
        return decorated_rowfactory
    return decorator

use it as:

@naming(rename=False, case=str.lower)
def rowfactory(cursor, row):
   casted = []
   ....
   ....
   return tuple(casted)

oracle.py

import cx_Oracle as Database
from cx_Oracle import *
import mybase

class Cursor(Database.Cursor):

    def execute(self, statement, args=None):
        prepareNested = (statement is not None and self.statement != statement)
        result = super(self.__class__, self).execute(statement, args or [])
        if prepareNested:
            if self.description:
                self.rowfactory = lambda *row: mybase.rowfactory(self, row)
        return result

    def close(self):
        try:
            super(self.__class__, self).close()
        except Database.InterfaceError:
            "already closed"

class Connection(Database.Connection):

    def cursor(self):
        Cursor(self)

connect = Connection

Now, instead of import cx_oracle import oracle in user script as:

user.py

import oracle

dsn = oracle.makedsn('HOSTNAME', 1521, service_name='dev_server')
db = connect('username', 'password', dsn)
cursor = db.cursor()
cursor.execute("""
  SELECT 'Grijesh' as FirstName, 
         'Chauhan' as LastName,
         CAST('10560.254' AS NUMBER(10, 2)) as Salary
  FROM DUAL
""")
row = cursor.fetchone()
print ("First Name is %s" % row.firstname) # => Grijesh
print ("Last Name is %s" % row.lastname) # => Chauhan
print ("Salary is %r" % row.salary) # => Decimal('10560.25')

Give it a Try!!

Auroraauroral

Answer 4

Building up on answer by @maelcum73 :

curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))

The issue with this solution is that you need to re-set this after every execution.

Going one step further, you can create a shell around the cursor object like so:

    class dictcur(object):
        # need to monkeypatch the built-in execute function to always return a dict
        def __init__(self, cursor):
            self._original_cursor = cursor

        def execute(self, *args, **kwargs):
            # rowfactory needs to be set AFTER EACH execution!
            self._original_cursor.execute(*args, **kwargs)
            self._original_cursor.rowfactory = lambda *a: dict(
                zip([d[0] for d in self._original_cursor.description], a)
            )
            # cx_Oracle's cursor's execute method returns a cursor object
            # -> return the correct cursor in the monkeypatched version as well!
            return self._original_cursor

        def __getattr__(self, attr):
            # anything other than the execute method: just go straight to the cursor
            return getattr(self._original_cursor, attr)
    dict_cursor = dictcur(cursor=conn.cursor())

Using this dict_cursor, every subsequent dict_cursor.execute() call will return a dictionary. Note: I tried monkeypatching the execute method directly, however that was not possible because it is a built-in method.

Damning