py查询结果如何转成json

Python 查询结果转 JSON 完整指南：从数据库到 API 的数据桥梁

在 Python 开发中，我们经常需要从各种数据源（如数据库、API、文件等）查询数据，并将这些查询结果转换为 JSON 格式，JSON（JavaScript Object Notation）因其轻量级、易读、易解析的特性，成为了数据交换的事实标准，本文将详细介绍如何在 Python 中将不同来源的查询结果转换为 JSON,涵盖从基础到进阶的各种场景和技巧。

为什么需要将查询结果转为 JSON？

在转换方法之前,我们先了解一下为什么需要这一步：

API 响应：许多现代 API 要求使用 JSON 格式进行数据交互，当你的 Python 应用作为后端服务时，需要将从数据库查询到的数据封装成 JSON 响应返回给前端。
数据存储：JSON 文件是一种简单且通用的数据存储方式，将查询结果保存为 JSON 文件，便于后续的数据分析、备份或迁移。
数据传输：在不同系统或组件之间传输数据时，JSON 格式具有良好的跨平台和跨语言兼容性。
配置管理：某些应用配置或中间数据可以以 JSON 形式存储和读取,查询结果可能直接用于此类配置。

准备工作：Python JSON 模块简介

Python 标准库中的 json 模块提供了强大的 JSON 处理功能，在开始转换之前,我们需要了解其核心方法：

json.dumps(obj): 将 Python 对象转换为 JSON 格式的字符串。
json.dump(obj, fp): 将 Python 对象转换为 JSON 格式字符串，并直接写入文件对象 fp。
json.loads(s): 将 JSON 格式的字符串 s 转换为 Python 对象。
json.load(fp): 从文件对象 fp 中读取 JSON 数据并转换为 Python 对象。

对于“查询结果转 JSON”，我们主要使用 json.dumps() 和 json.dump()。

常见查询结果类型转 JSON 的方法

查询结果的来源多种多样，其初始的 Python 数据类型也不同,以下是几种常见场景的转换方法。

从数据库查询结果（如 SQLite, MySQL, PostgreSQL）转 JSON

使用数据库 API（如 sqlite3, pymysql, psycopg2）查询数据后，通常得到的结果是元组（tuple）列表，或者包含列名信息的字典列表（如果使用了游标的方法如 fetchall() 并配合 description）。

方法 1：直接转换元组列表

如果查询结果是简单的元组列表，json.dumps() 默认可以处理，但键名会是数字索引（JSON 数组索引）。

import sqlite3
import json
# 1. 连接数据库并查询
conn = sqlite3.connect(':memory:') # 使用内存数据库示例
cursor = conn.cursor()
cursor.execute("CREATE TABLE users (id INTEGER, name TEXT, email TEXT)")
cursor.execute("INSERT INTO users VALUES (1, 'Alice', 'alice@example.com')")
cursor.execute("INSERT INTO users VALUES (2, 'Bob', 'bob@example.com')")
conn.commit()
# 2. 执行查询，获取元组列表
cursor.execute("SELECT * FROM users")
query_results = cursor.fetchall() # 结果: [(1, 'Alice', 'alice@example.com'), (2, 'Bob', 'bob@example.com')]
# 3. 转换为 JSON 字符串
json_string = json.dumps(query_results, indent=4)
print("元组列表转 JSON:")
print(json_string)
# 4. 可选：写入 JSON 文件
with open('users_tuple.json', 'w') as f:
    json.dump(query_results, f, indent=4)
cursor.close()
conn.close()

输出：

[
    [
        1,
        "Alice",
        "alice@example.com"
    ],
    [
        2,
        "Bob",
        "bob@example.com"
    ]
]

方法 2：转换为字典列表（推荐）

这种方式更直观，JSON 的键为列名。

import sqlite3
import json
# (数据库连接和插入数据同上...)
cursor.execute("SELECT * FROM users")
# 获取列名
columns = [description[0] for description in cursor.description]
# 获取数据行
rows = cursor.fetchall()
# 将元组列表转换为字典列表
query_results_dict = [dict(zip(columns, row)) for row in rows]
# 3. 转换为 JSON 字符串
json_string_dict = json.dumps(query_results_dict, indent=4)
print("\n字典列表转 JSON:")
print(json_string_dict)
# 4. 可选：写入 JSON 文件
with open('users_dict.json', 'w') as f:
    json.dump(query_results_dict, f, indent=4)
cursor.close()
conn.close()

输出：

[
    {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com"
    },
    {
        "id": 2,
        "name": "Bob",
        "email": "bob@example.com"
    }
]

提示：许多现代数据库驱动（如 SQLAlchemy ORM）可以直接返回字典列表,简化了这一过程。

从 API 响应中解析数据（本身已是 JSON 或需处理）

如果查询是通过 API 获取的，响应体通常是 JSON 字符串或可以直接解析为 Python 对象。

import json
import requests # 需要安装 requests 库
# 假设这是一个返回 JSON 的 API 端点
api_url = "https://jsonplaceholder.typicode.com/users"
try:
    # 1. 发送 GET 请求
    response = requests.get(api_url)
    response.raise_for_status()  # 如果请求失败则抛出异常
    # 2. response.json() 直接将响应体解析为 Python 列表/字典
    api_query_results = response.json()
    print("API 返回的 Python 对象类型:", type(api_query_results))
    # 输出: API 返回的 Python 对象类型: <class 'list'>
    # 3. 如果需要重新格式化为 JSON 字符串（例如美化或传输）
    json_string_from_api = json.dumps(api_query_results, indent=4)
    # print(json_string_from_api)
    # 4. 写入文件
    with open('api_users.json', 'w') as f:
        json.dump(api_query_results, f, indent=4)
except requests.exceptions.RequestException as e:
    print(f"API 请求失败: {e}")
except json.JSONDecodeError as e:
    print(f"JSON 解析失败: {e}")

从其他数据结构（如列表、字典）转 JSON

有时查询结果可能不是来自数据库或 API，而是程序内部生成的列表、字典等结构，这些可以直接使用 json.dumps() 转换。

import json
# 示例：自定义查询结果（模拟）
custom_query_results = [
    {"product_id": "P001", "product_name": "Laptop", "price": 1200.99, "in_stock": True},
    {"product_id": "P002", "product_name": "Mouse", "price": 25.50, "in_stock": False}
]
# 直接转换为 JSON 字符串
json_string_custom = json.dumps(custom_query_results, indent=4)
print("自定义数据结构转 JSON:")
print(json_string_custom)
# 写入文件
with open('products.json', 'w') as f:
    json.dump(custom_query_results, f, indent=4)

处理转换中的常见问题与进阶技巧

处理非 JSON 原生类型（如 datetime, Decimal） json.dumps() 默认只能处理基本类型（str, int, float, bool, list, dict, None），如果查询结果中包含 datetime.date, datetime.datetime, decimal.Decimal 等类型，会抛出 TypeError。

解决方案：自定义 default 参数 可以通过 json.dumps() 的 default 参数指定一个函数,用于处理非原生类型。

import json
from datetime import datetime
from decimal import Decimal
data_with_custom_types = {
    "timestamp": datetime.now(),
    "price": Decimal("99.99"),
    "name": "Test Product"
}
def custom_serializer(obj):
    if isinstance(obj, datetime):
        return obj.isoformat() # 将 datetime 转换为 ISO 格式字符串
    elif isinstance(obj, Decimal):
        return float(obj) # 将 Decimal 转换为 float
    # 可以添加更多类型的处理
    raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
json_string_custom_types = json.dumps(data_with_custom_types, default=custom_serializer, indent=4)
print("\n处理非原生类型后的 JSON:")
print(json_string_custom_types)

输出：