使用python作为字典从postgresql查询

Question

我正在使用Python 2.7和postgresql 9.1.试图从查询中获取字典,我已经尝试了这里描述的代码:http: //wiki.postgresql.org/wiki/Using_psycopg2_with_PostgreSQL

import psycopg2
import psycopg2.extras
conn = psycopg2.connect("dbname=mydb host=localhost user=user password=password")
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute ("select * from port")
type(cur.fetchall())

它打印下一个答案:

<type 'list'>

打印项目本身,告诉我它是列表.除外答案是字典.

编辑:

尝试下一个:

ans = cur.fetchall()[0]
print ans
print type(ans)

回报

[288, 'T', 51, 1, 1, '192.168.39.188']
<type 'list'>

Answer 1

Tnx很多Andrey Shokhin,

完整的答案是:

#!/var/bin/python 
import psycopg2
import psycopg2.extras
conn = psycopg2.connect("dbname=uniart4_pr host=localhost user=user password=password")
cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute ("select * from port")
ans =cur.fetchall()
ans1 = []
for row in ans:
    ans1.append(dict(row))

print ans1  #actually it's return

Answer 2

这是正常的:当你调用.fetchall()方法返回元组列表.但如果你写

type(cur.fetchone())

它只返回一个类型的元组:

<class 'psycopg2.extras.DictRow'>

在此之后,您可以将其用作列表或类似字典:

cur.execute('SELECT id, msg FROM table;')
rec = cur.fetchone()
print rec[0], rec['msg']

您还可以使用简单的游标迭代器:

res = [json.dumps(dict(record)) for record in cursor] # it calls .fetchone() in loop

Answer 3

有一个内置的解决方案可以将结果作为字典集合获取：

from psycopg2.extras import RealDictCursor
cur = conn.cursor(cursor_factory=RealDictCursor)

修改自： https: //www.peterbe.com/plog/from-postgres-to-json-strings，版权所有 2013 Peter Bengtsson

Answer 4

也许我们可以进一步优化它

#!/var/bin/python 
import psycopg2
import psycopg2.extras

def get_dict_resultset(sql):
    conn = psycopg2.connect("dbname=pem host=localhost user=postgres password=Drupal#1008")
    cur = conn.cursor(cursor_factory=psycopg2.extras.DictCursor)
    cur.execute (sql)
    ans =cur.fetchall()
    dict_result = []
    for row in ans:
        dict_result.append(dict(row))
    return dict_result

sql = """select * from tablename"""
return get_dict_resultset(sql)

Answer 5

除了仅将查询结果作为字典列表返回之外，我还建议返回键值对（列名：行值）。这是我的建议：

import psycopg2
import psycopg2.extras

    conn = None
    try:
        conn = psycopg2.connect("dbname=uniart4_pr host=localhost user=user password=password")

        with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
            cursor.execute("SELECT * FROM table")
            column_names = [desc[0] for desc in cursor.description]
            res = cursor.fetchall()
            cursor.close()
        return map(lambda x: dict(zip(column_names, x)), res))

    except (Exception, psycopg2.DatabaseError) as e:
        logger.error(e)
    finally:
        if conn is not None:
            conn.close()

Answer 6

如果您不想使用 apsycopg2.extras.DictCursor您可以使用cursor.description以下命令为结果创建字典列表：

# connect
connection = psycopg2.connect()
cursor = connection.cursor()

# query
cursor.execute("SELECT * FROM myTable")

# transform result
columns = list(cursor.description)
result = cursor.fetchall()

# make dict
results = []
for row in result:
    row_dict = {}
    for i, col in enumerate(columns):
        row_dict[col.name] = row[i]
    results.append(row_dict)

# display
print(result)

我经常使用以下功能：

def select_query_dict(connection, query, data=[]):
    """
    Run generic select query on db, returns a list of dictionaries
    """
    logger.debug('Running query: {}'.format(query))

    # Open a cursor to perform database operations
    cursor = connection.cursor()
    logging.debug('Db connection succesful')

    # execute the query
    try:
        logger.info('Running query.')
        if len(data):
            cursor.execute(query, data)
        else:
            cursor.execute(query)
        columns = list(cursor.description)
        result = cursor.fetchall()
        logging.debug('Query executed succesfully')
    except (Exception, psycopg2.DatabaseError) as e:
        logging.error(e)
        cursor.close()
        exit(1)

    cursor.close()

    # make dict
    results = []
    for row in result:
        row_dict = {}
        for i, col in enumerate(columns):
            row_dict[col.name] = row[i]
        results.append(row_dict)

    return results