甲骨文：我怎样才能把每一行作为字典接收呢？

Question

默认情况下，cx_Oracle将每一行作为元组返回。

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

如何将每一行作为字典返回？

Answer 1

^{老问题，但添加一些与Python配方相关的有用链接}

根据cx_Oracle文档：

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.

cx_Oracle - Python Interface for Oracle Database还指向GitHub存储库以获取许多有用的sample示例。请检查GenericRowFactory.py。

谷歌：这个PPT可以进一步帮助：^[PDF]CON6543 Python and Oracle Database - RainFocus

配方

Django数据库后台为Oracle under the hood使用的是cx_Oracle。在早期的版本（Django 1.11-）中，他们编写了^{}，还将cx_Oracle的数字数据类型转换为相关的Python数据，并将字符串转换为unicode。

如果您安装了Django，请按以下方式检查base.py：

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

可以从$DJANGO_DIR/db/backends/oracle/base.py中借用_rowfactory()，并可以在decorator naming下面应用，使其返回namedtuple，而不是简单的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

用作：

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

甲骨文.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

现在，不要在用户脚本中导入cx_oracle导入oracle：

用户.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')

试试看！！

Answer 2

非常简短的版本：

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

在Python 3.7.0和cx上测试Oracle 7.1.2

Answer 3

您可以重写游标的rowfactory方法。每次执行查询时都需要这样做。

这是标准查询的结果，一个元组。

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

返回命名元组：

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')

返回词典：

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}

阿默里·福盖特·德·阿尔克的功劳： http://sourceforge.net/p/cx-oracle/mailman/message/27145597