如何将 pyodbc 游标输出（来自或.fetchone）序列化为Python 字典？.fetchmany.fetchall
我正在使用bottlepy并且需要返回dict，以便它可以将它作为JSON返回。
问问题
121402 次
10 回答
190
如果您不提前知道列，请使用Cursor.description构建列名列表，并使用每一行压缩以生成字典列表。示例假设已建立连接和查询：
>>> cursor = connection.cursor().execute(sql)
>>> columns = [column[0] for column in cursor.description]
>>> print(columns)
['name', 'create_date']
>>> results = []
>>> for row in cursor.fetchall():
... results.append(dict(zip(columns, row)))
...
>>> print(results)
[{'create_date': datetime.datetime(2003, 4, 8, 9, 13, 36, 390000), 'name': u'master'},
{'create_date': datetime.datetime(2013, 1, 30, 12, 31, 40, 340000), 'name': u'tempdb'},
{'create_date': datetime.datetime(2003, 4, 8, 9, 13, 36, 390000), 'name': u'model'},
{'create_date': datetime.datetime(2010, 4, 2, 17, 35, 8, 970000), 'name': u'msdb'}]
于 2013-05-13T13:22:23.650 回答
14
将@Beargle 的结果与bottlepy 一起使用，我能够创建这个非常简洁的查询暴露端点：
@route('/api/query/<query_str>')
def query(query_str):
cursor.execute(query_str)
return {'results':
[dict(zip([column[0] for column in cursor.description], row))
for row in cursor.fetchall()]}
于 2013-05-14T12:37:14.300 回答
7
这是您可以使用的简短版本
>>> cursor.select("<your SQL here>")
>>> single_row = dict(zip(zip(*cursor.description)[0], cursor.fetchone()))
>>> multiple_rows = [dict(zip(zip(*cursor.description)[0], row)) for row in cursor.fetchall()]
正如您可能知道的那样，当您将 * 添加到列表时，您基本上会删除列表，将各个列表条目作为参数留给您正在调用的函数。通过使用 zip，我们选择第 1 到第 n 个条目并将它们拉到一起，就像裤子上的拉链一样。
所以通过使用
zip(*[(a,1,2),(b,1,2)])
# interpreted by python as zip((a,1,2),(b,1,2))
你得到
[('a', 'b'), (1, 1), (2, 2)]
由于 description 是一个带有元组的元组，其中每个元组描述了每列的标题和数据类型，因此您可以提取每个元组的第一个
>>> columns = zip(*cursor.description)[0]
相当于
>>> columns = [column[0] for column in cursor.description]
于 2014-12-11T12:00:57.323 回答
6
对于游标不可用的情况 - 例如，当某些函数调用或内部方法返回了行时，您仍然可以使用 row.cursor_description 创建字典表示
def row_to_dict(row):
return dict(zip([t[0] for t in row.cursor_description], row))
于 2019-02-17T08:14:55.583 回答
3
主要是关闭@Torxed 响应，我创建了一套完整的通用函数来将模式和数据查找到字典中：
def schema_dict(cursor):
cursor.execute("SELECT sys.objects.name, sys.columns.name FROM sys.objects INNER JOIN sys.columns ON sys.objects.object_id = sys.columns. object_id WHERE sys.objects.type = 'U';")
schema = {}
for it in cursor.fetchall():
if it[0] not in schema:
schema[it[0]]={'scheme':[]}
else:
schema[it[0]]['scheme'].append(it[1])
return schema
def populate_dict(cursor, schema):
for i in schema.keys():
cursor.execute("select * from {table};".format(table=i))
for row in cursor.fetchall():
colindex = 0
for col in schema[i]['scheme']:
if not 'data' in schema[i]:
schema[i]['data']=[]
schema[i]['data'].append(row[colindex])
colindex += 1
return schema
def database_to_dict():
cursor = connect()
schema = populate_dict(cursor, schema_dict(cursor))
随意在此上进行所有代码高尔夫以减少行数；但与此同时，它有效！
;)
于 2013-05-13T18:18:24.113 回答
1
我需要的，这与 OP 所要求的略有不同：
如果您想完全概括执行 SQL 选择查询的例程，但您需要通过索引号而不是名称来引用结果，您可以这样做，使用列表列表而不是字典。
每行返回的数据在返回列表中表示为字段（列）值的列表。
列名可以作为返回列表的第一项提供，因此在调用例程中解析返回的列表可以非常简单和灵活。
这样，执行数据库调用的例程就不需要知道它正在处理的数据的任何信息。这是这样一个例程：
def read_DB_Records(self, tablename, fieldlist, wherefield, wherevalue) -> list:
DBfile = 'C:/DATA/MyDatabase.accdb'
# this connection string is for Access 2007, 2010 or later .accdb files
conn = pyodbc.connect(r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};DBQ='+DBfile)
cursor = conn.cursor()
# Build the SQL Query string using the passed-in field list:
SQL = "SELECT "
for i in range(0, len(fieldlist)):
SQL = SQL + "[" + fieldlist[i] + "]"
if i < (len(fieldlist)-1):
SQL = SQL + ", "
SQL = SQL + " FROM " + tablename
# Support an optional WHERE clause:
if wherefield != "" and wherevalue != "" :
SQL = SQL + " WHERE [" + wherefield + "] = " + "'" + wherevalue + "';"
results = [] # Create the results list object
cursor.execute(SQL) # Execute the Query
# (Optional) Get a list of the column names returned from the query:
columns = [column[0] for column in cursor.description]
results.append(columns) # append the column names to the return list
# Now add each row as a list of column data to the results list
for row in cursor.fetchall(): # iterate over the cursor
results.append(list(row)) # add the row as a list to the list of lists
cursor.close() # close the cursor
conn.close() # close the DB connection
return results # return the list of lists
于 2018-02-22T18:29:50.690 回答
1
我喜欢@bryan 和@foo-stack 的答案。如果您正在使用 postgresql 并且正在使用，则psycopg2可以使用psycopg2 中的一些好东西DictCursor通过在从连接创建游标时将 cursorfactory 指定为 a 来实现相同的目的，如下所示：
cur = conn.cursor( cursor_factory=psycopg2.extras.DictCursor )
所以现在你可以执行你的 sql 查询，你会得到一个字典来获取你的结果，而不需要手动映射它们。
cur.execute( sql_query )
results = cur.fetchall()
for row in results:
print row['row_no']
请注意，您必须这样import psycopg2.extras做才能正常工作。
于 2018-03-01T10:06:55.507 回答
1
步骤如下：
- 导入库：
from pandas import DataFrame
import pyodbc
import sqlalchemy
- 从本地数据库获取结果：
db_file = r'xxx.accdb'
odbc_conn_str = 'DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=%s' % (db_file)
conn = pyodbc.connect(odbc_conn_str)
cur = conn.cursor()
qry = cur.execute("SELECT * FROM tbl")
columns = [column[0] for column in cur.description]
results = []
for row in cur.fetchall():
results.append(dict(zip(columns, row)))
df = DataFrame(results)
df
于 2021-04-10T21:25:02.630 回答
0
假设你知道你的列名！此外，这里有三种不同的解决方案，
您可能想看看最后一个！
colnames = ['city', 'area', 'street']
data = {}
counter = 0
for row in x.fetchall():
if not counter in data:
data[counter] = {}
colcounter = 0
for colname in colnames:
data[counter][colname] = row[colcounter]
colcounter += 1
counter += 1
这是一个索引版本，不是最漂亮的解决方案，但它会起作用。另一种方法是将列名索引为字典键，每个键中的列表包含按行号顺序排列的数据。通过做：
colnames = ['city', 'area', 'street']
data = {}
for row in x.fetchall():
colindex = 0
for col in colnames:
if not col in data:
data[col] = []
data[col].append(row[colindex])
colindex += 1
写这个，我知道做for col in colnames可以被替换，for colindex in range(0, len())但你明白了。后面的示例在不获取所有数据但一次获取一行时很有用，例如：
对每一行数据使用dict
def fetchone_dict(stuff):
colnames = ['city', 'area', 'street']
data = {}
for colindex in range(0, colnames):
data[colnames[colindex]] = stuff[colindex]
return data
row = x.fetchone()
print fetchone_dict(row)['city']
获取表名（我认为.. 感谢 Foo Stack）：来自下面的小熊
更直接的解决方案！
cursor.execute("SELECT sys.objects.name, sys.columns.name FROM sys.objects INNER JOIN sys.columns ON sys.objects.object_id = sys.columns. object_id WHERE sys.objects.type = 'U';")
schema = {}
for it in cursor.fetchall():
if it[0] in schema:
schema[it[0]].append(it[1])
else:
schema[it[0]] = [it[1]]
于 2013-05-13T11:04:33.560 回答
0
我知道它很旧，我只是在重述其他人已经说过的话。但我发现这种方式很整洁，因为它也可以安全注射。
def to_dict(row):
return dict(zip([t[0] for t in row.cursor_description], row))
def query(cursor, query, params=[], cursor_func=to_dict):
cursor.execute(query, params)
results = [cursor_func(row) for row in cursor.fetchall()]
return results
quotes = query(cursor, "select * from currency where abbreviation like ?", ["USD"])
于 2021-04-08T13:51:40.427 回答