Python将MySQL查询结果转换为JSON

Question

我正在尝试实现REST API，其中的一部分是将数据格式化为JSON。我可以从MySQL数据库中检索数据，但是我收到的对象与我的期望不符。这是我的代码：

from flask import Flask
from flask.ext.mysqldb import MySQL

app = Flask(__name__)
app.config['MYSQL_HOST'] = '127.0.0.1'
app.config['MYSQL_USER'] = 'root'
app.config['MYSQL_PASSWORD'] = 'password'
app.config['MYSQL_DB'] = 'hello_db'
mysql = MySQL(app)

@app.route('/hello')
def index():
   cur = mysql.connection.cursor()
   cur.execute('''SELECT * FROM Users WHERE id=1''')
   rv = cur.fetchall()
   return str(rv)

if __name__ == '__main__':
   app.run(debug=True)

结果:

((1L, u'my_username', u'my_password'),)

我要如何实现返回以下JSON格式：

{
 "id":1, 
 "username":"my_username", 
 "password":"my_password"
}

- edmamerto

Answer 1

您可以使用光标描述来提取行标题：row_headers=[x[0] for x in cursor.description]在执行语句之后。然后，您可以将其与SQL结果进行压缩以生成JSON数据。 因此，您的代码将类似于：

from flask import Flask
from flask.ext.mysqldb import MySQL
import json
app = Flask(__name__)
app.config['MYSQL_HOST'] = '127.0.0.1'
app.config['MYSQL_USER'] = 'root'
app.config['MYSQL_PASSWORD'] = 'password'
app.config['MYSQL_DB'] = 'hello_db'
mysql = MySQL(app)

@app.route('/hello')
def index():
   cur = mysql.connection.cursor()
   cur.execute('''SELECT * FROM Users WHERE id=1''')
   row_headers=[x[0] for x in cur.description] #this will extract row headers
   rv = cur.fetchall()
   json_data=[]
   for result in rv:
        json_data.append(dict(zip(row_headers,result)))
   return json.dumps(json_data)

if __name__ == '__main__':
   app.run(debug=True)

在return语句中，您可以使用jsonify代替评论中建议的json.dumps。

- Mani

---

我收到了 Decimal('0.00') is not JSON serializable 错误。 - Ashok Sri

你的MySQL结果中可能有十进制数。需要将Decimal对象转换为sfloat。试试这个 - Mani

“return jsonify(json_data)”这种写法相比于“return json.dumps(json_data)”更加简洁，你觉得呢？ - Ri1a

我认为jsonify是外部库。考虑尝试使用默认库。 - Mani

你是对的，但这里是 Flask Docs API JSON Support 中jsonify()的引语：“此函数包装了dumps()，并添加了一些增强功能以使生活更轻松。它将JSON输出转换为带有应用程序/json mimetype的响应对象。” - Ri1a

程序按预期工作，因为它在响应中给出了一个JSON数组，但请确保使用更新的导入，因为.ext格式已过时。这是您需要添加的内容：from flask_mysqldb import MySQL 如果您是初学者，请不要忘记将flask_mysqldb添加到您的要求文件中或执行pip install flask_mysqldb。 - Akhil

Answer 2

或许有一种更简单的方法来做到这一点：返回一个字典并将其转换为JSON。

只需按照MySQL文档中提到的方式，将dictionary=True传递给游标构造函数即可。

import json
import mysql.connector

db = mysql.connector.connect(host='127.0.0.1',
                             user='admin',
                             passwd='password',
                             db='database',
                             port=3306)

# This is the line that you need
cursor = db.cursor(dictionary=True)

name = "Bob"
cursor.execute("SELECT fname, lname FROM table WHERE fname=%s;", (name))

result = cursor.fetchall()

print(f"json: {json.dumps(result)}")

这将打印 -

json: [{'fname': "Bob", 'lname': "Dole"}, {'fname': "Bob", 'lname': "Marley"}]

（假设这些Bobs在表中。）

请注意，通过这种方式保留了类型，这是一件好事，但需要将其转换、解析或序列化为字符串；例如，如果有一个日期，SQL查询可能会返回一个datetime对象，这将根据您的下一步骤进行解析或序列化。一个很好的序列化方法可以参考this answer。

- NonCreature0714

Answer 3

从你的输出来看，似乎是得到了一个元组？如果是这样，那么你应该可以直接使用map函数。

from flask import Flask, jsonify
from flask.ext.mysqldb import MySQL

app = Flask(__name__)
app.config['MYSQL_HOST'] = '127.0.0.1'
app.config['MYSQL_USER'] = 'root'
app.config['MYSQL_PASSWORD'] = 'password'
app.config['MYSQL_DB'] = 'hello_db'
mysql = MySQL(app)

@app.route('/hello')
def index():
   cur = mysql.connection.cursor()
   cur.execute('''SELECT * FROM Users WHERE id=1''')
   rv = cur.fetchall()
   payload = []
   content = {}
   for result in rv:
       content = {'id': result[0], 'username': result[1], 'password': result[2]}
       payload.append(content)
       content = {}
   return jsonify(payload)

if __name__ == '__main__':
   app.run(debug=True)

- Mike Tung

---

这不是一个适合期望动态响应的人的正确方法。 - Akhil

Answer 4

我是如何通过设置app.config['MYSQL_CURSORCLASS']和使用jsonify来解决这个问题的。

from flask import Flask, jsonify
from flask_mysqldb import MySQL

app = Flask(__name__)
app.config['MYSQL_HOST'] = '127.0.0.1'
app.config['MYSQL_USER'] = 'root'
app.config['MYSQL_PASSWORD'] = 'password'
app.config['MYSQL_DB'] = 'hello_db'
app.config['MYSQL_CURSORCLASS'] = 'DictCursor' # line of code you require

mysql = MySQL(app)

@app.route('/hello')
def index():
   cur = mysql.connection.cursor()
   cur.execute("SELECT * FROM Users WHERE id=1")
   rv = cur.fetchall()
   return jsonify(rv) # use jsonify here

if __name__ == '__main__':
   app.run(debug=True)

- Hashan Shalitha

Answer 5

对于使用Django的用户，您可以从django.http中导入JsonResponse来完成您的工作。

示例代码片段：

from django.http import JsonResponse
from django.db import connection

def home(request):
    with connection.cursor() as cursor:
        cursor.execute("select * from YOUR_TABLE")
        columns = [col[0] for col in cursor.description]
        return JsonResponse([
            dict(zip(columns, row))
            for row in cursor.fetchall()
        ], safe=False)

注意：已添加safe=False以将字典列表转换为Json。

- Nishit Doshi

Answer 6

如果您想将更多的 select 查询结果 转换成 JSON 文件，下面这段简单的程序代码就能够胜任。欲了解 更多 详情，请前往github，有三种解决方案可供选择。

鉴于已经对许多问题进行了澄清，现在简单介绍一下这些方法：

• class DateTimeEncoder(JSONEncoder) - 编码器以支持日期时间 - doc。
• get_current_date_time - 当前时间用于区分是否使用可变数据
• query_db - 使用游标描述提取行标题，您将获得一个字典对象标题：值的数组。
• write_json(query_path) - 读取 SQL 并在已创建的 output 文件夹中生成 JSON
• convertion_mysql - 使用 glob 查找具有扩展名 .sql 的目录中的所有文件，并调用所描述和定义的方法 write_json

import json
from json import JSONEncoder
import datetime
import os
import glob

class DateTimeEncoder(JSONEncoder):
        def default(self, obj):
            if isinstance(obj, (datetime.date, datetime.datetime)):
                return obj.isoformat()

def get_current_date_time():
    return datetime.datetime.now().strftime('%a_%d.%b.%Y_%H.%M.%S')

def query_db(query):
    cur = mysql.connection.cursor()
    cur.execute(query)
    r = [dict((cur.description[i][0], value) for i, value in enumerate(row)) for row in cur.fetchall()]
    cur.connection.close()
    return r

def write_json(query_path):
    with open(f"{query_path}", 'r') as f:
        sql = f.read().replace('\n', ' ')
    file_name = os.path.splitext(os.path.basename(f"{query_path}"))[0]
    with open(f"../output/{file_name}_{get_current_date_time()}.json", 'w', encoding='utf-8') as f:
        json.dump(query_db(sql), f, ensure_ascii=False, indent=4, cls=DateTimeEncoder)

def convertion_mysql():
    mysql_query = [f for f in glob.glob("../myqls/*.sql")]
    for sql in mysql_query:
        write_json(sql)

if __name__ == "__main__":
    convertion_mysql()

- Milovan Tomašević