cx_Oracle: 如何将每一行作为字典接收？
cx_Oracle: How can I receive each row as a dictionary?
默认情况下，cx_Oracle return将每一行作为一个元组。
>>> import cx_Oracle
>>> conn=cx_Oracle.connect('scott/tiger')
>>> curs=conn.cursor()
>>> curs.execute("select * from foo");
>>> curs.fetchone()
(33, 'blue')
如何 return 将每一行作为字典？
您可以覆盖游标的 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}
感谢 Amaury Forgeot d'Arc：
http://sourceforge.net/p/cx-oracle/mailman/message/27145597
老问题，但添加了一些有用的链接 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 Also points to GitHub repository for lots of helpful sample examples. Please check GenericRowFactory.py。
Google 搜索：此 PPT 可以进一步提供帮助：[PDF]CON6543 Python and Oracle Database - RainFocus
食谱
Oracle 的 Django 数据库后台使用 cx_Oracle。在早期版本（Django 1.11-）中，他们编写了 _rowfactory(cursor, row) 这也将 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() 并可以应用下面的装饰器 naming 使其成为 return namedtuple 而不是简单的 tuple .
我的base.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)
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
现在，代替 import cx_oracle 在用户脚本中将 oracle 导入为：
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')
试一试！！
一个非常简短的版本：
curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))
在 Python 3.7.0 和 cx_Oracle 7.1.2
上测试
以@maelcum73 的回答为基础：
curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))
此解决方案的问题是您需要在每次执行后重新设置。
更进一步，您可以像这样在光标对象周围创建一个 shell：
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())
使用此 dict_cursor，每个后续 dict_cursor.execute() 调用都会 return 字典。注意：我尝试直接对 execute 方法进行 monkeypatching，但是这是不可能的，因为它是一个内置方法。
默认情况下，cx_Oracle return将每一行作为一个元组。
>>> import cx_Oracle
>>> conn=cx_Oracle.connect('scott/tiger')
>>> curs=conn.cursor()
>>> curs.execute("select * from foo");
>>> curs.fetchone()
(33, 'blue')
如何 return 将每一行作为字典？
您可以覆盖游标的 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}
感谢 Amaury Forgeot d'Arc： http://sourceforge.net/p/cx-oracle/mailman/message/27145597
老问题，但添加了一些有用的链接 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 Also points to GitHub repository for lots of helpful sample examples. Please check GenericRowFactory.py。
Google 搜索：此 PPT 可以进一步提供帮助：[PDF]CON6543 Python and Oracle Database - RainFocus
食谱
Oracle 的 Django 数据库后台使用 cx_Oracle。在早期版本（Django 1.11-）中，他们编写了 _rowfactory(cursor, row) 这也将 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() 并可以应用下面的装饰器 naming 使其成为 return namedtuple 而不是简单的 tuple .
我的base.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)
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
现在，代替 import cx_oracle 在用户脚本中将 oracle 导入为：
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')
试一试！！
一个非常简短的版本：
curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))
在 Python 3.7.0 和 cx_Oracle 7.1.2
上测试以@maelcum73 的回答为基础：
curs.rowfactory = lambda *args: dict(zip([d[0] for d in curs.description], args))
此解决方案的问题是您需要在每次执行后重新设置。
更进一步，您可以像这样在光标对象周围创建一个 shell：
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())
使用此 dict_cursor，每个后续 dict_cursor.execute() 调用都会 return 字典。注意：我尝试直接对 execute 方法进行 monkeypatching，但是这是不可能的，因为它是一个内置方法。