从 Python 中的 Sqlite 查询发送 JSON 响应答案

【问题标题】：Send JSON response from Sqlite queries in Python从 Python 中的 Sqlite 查询发送 JSON 响应
【发布时间】：2014-10-23 05:40:43
【问题描述】：

我有一个 Python http 服务器，它侦听基于 JSON 的请求。收到请求后，从 JSON 输入中解析 key，并查询具有该 key 的 Sqlite 数据库。现在我想用结果 JSON 消息响应请求。我是 Python 新手，不知道如何操作。

我的代码结构如下：

 import ...

 key=...;//get key from request
 con = lite.connect('test.db')
 with con:
    con.row_factory = lite.Row
    cur = con.cursor()
    cur.execute("SELECT * FROM mytable ");
    while True:

        row = cur.fetchone()

        if row == None:
            break
        if key==row['key']:
            # How can I add the record to the response?

并且处理程序会这样写响应（在一个类中继承 BaseHTTPRequestHandler 并由一个线程启动）

self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(??????) # What do I need to write here?

【问题讨论】：

导入json； self.wfile.write(json.dumps(yourdict))
@gosom 如何从 sqlite 结果/行中组织这样的字典？我使用 LocalData.records[row[0]]=row 存储每个匹配的行，并使用 self.wfile.write(json.dumps(LocalData.records)) 写入输出。它说“...不是 JSON 可序列化的”。

标签： python json sqlite

【解决方案1】：

返回 JSON 响应就这么简单：

import json
import sqlite3

def get_my_jsonified_data(key):
    with sqlite3.connect('test.db') as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT * FROM mytable WHERE column=?;", [key])
        data = cursor.fetchall()
        return json.dumps(data)

（假设lite 是sqlite3 的别名）

注意其他几点：

我已经删除了while True: 循环。它效率极低、不安全且难以阅读；
我在 SQL 查询中添加了对 key 的检查（为什么还要从 DB 加载不必要的数据？）

【讨论】：

谢谢。对于检查密钥，这是我的错，因为我正在逐个添加代码，因此显然没有必要（只需使用 sql）。另一个问题是，如果结果很大，fetchall() 可能会出现问题（内存、响应大小等）？
@mrmoment 除非您正在进行流式传输（看起来不像那样），否则无论如何您都必须将所有数据加载到内存中。如果您正在进行流式传输，请使用 cursor.fetchamany(10000) 例如。逐项获取非常低效。
知道了。我想我可以使用“limit”子句来减少 fetchall() 的结果。再次感谢。
当我在 Python3 中尝试这个时，我在 json.dumps(data) 上得到 TypeError: Object of type 'Row' is not JSON serializable。还需要配置什么吗？
@PrahladYeri 看来您在某处使用conn.row_factory = sqlite3.Row。在这种情况下，您可以尝试json.dumps([tuple(row) for row in data])。

【解决方案2】：

你可以试试这个

import sqlite3


def row_to_dict(cursor: sqlite3.Cursor, row: sqlite3.Row) -> dict:
    data = {}
    for idx, col in enumerate(cursor.description):
        data[col[0]] = row[idx]
    return data

with sqlite3.connect(db_path) as con:
    con.row_factory = row_to_dict
    result = con.execute('SELECT * FROM table_name')
    print(result.fetchall())

【讨论】：