PyMySQL JSON 数据处理：Python 与 MySQL 交互

PyMySQL 的类型转换逻辑主要在 converters.py 中实现。该模块定义了 Python 数据类型与 MySQL 数据类型之间的转换规则，确保字典、列表等复杂结构能正确序列化为 JSON 格式存储到数据库中。

数据类型映射

PyMySQL 处理 JSON 数据时，会将 Python 数据类型按照以下规则映射为 MySQL JSON 类型：

这些转换规则由 converters.py 中的 encoders 字典定义，确保 Python 数据能正确序列化为 MySQL 可接受的格式。

JSON 数据查询与解析

从 MySQL 查询 JSON 数据时，PyMySQL 会自动将 JSON 字段解析为 Python 字典或列表，使你可以像操作普通 Python 数据结构一样处理 JSON 数据。

基本查询操作

import pymysql
from pymysql.cursors import DictCursor

connection = pymysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    db='test_db',
    charset='utf8mb4',
    cursorclass=DictCursor
)

try:
    with connection.cursor() as cursor:
        # 查询 JSON 数据
        select_sql = "SELECT id, name, preferences FROM user_profiles WHERE id = %s"
        cursor.execute(select_sql, (1,))
        result = cursor.fetchone()

        if result:
            print(f"User: {result['name']}")
            print("Preferences:")
            
            # 直接操作 JSON 字段解析后的 Python 字典
            preferences = result['preferences']
            
            # 访问 JSON 对象属性
            print(f"Theme: {preferences['theme']}")
            
            # 访问嵌套 JSON 对象
            print(f"Email notifications: {preferences['notifications']['email']}")
            
            # 遍历 JSON 数组
            print("Favorite features:")
            for feature in preferences['favorite_features']:
                print(f" - {feature}")
            
            # 修改 JSON 数据
            preferences['theme'] = 'light'
            preferences['display_options']['font_size'] = 16
            
            # 更新修改后的 JSON 数据
            update_sql = "UPDATE user_profiles SET preferences = %s WHERE id = %s"
            cursor.execute(update_sql, (preferences, result['id']))
            connection.commit()
finally:
    connection.close()

当从 MySQL 读取 JSON 字段时，PyMySQL 会使用 converters.py 中的 decoders 字典定义的规则，将 JSON 数据转换回相应的 Python 数据类型。这使得我们可以直接像操作普通 Python 字典一样处理 JSON 数据。

使用 MySQL JSON 函数查询

MySQL 提供了丰富的 JSON 函数，可以在 SQL 层面直接操作 JSON 数据。结合 PyMySQL，我们可以灵活地查询和修改 JSON 字段的特定部分：

-- 获取所有用户的主题偏好
SELECT name, preferences->'$.theme' AS theme FROM user_profiles;

-- 使用 JSON_EXTRACT 获取嵌套属性
SELECT name, JSON_EXTRACT(preferences, '$.display_options.font_size') AS font_size 
FROM user_profiles 
WHERE JSON_EXTRACT(preferences, '$.display_options.font_size') > 14;

-- 使用 JSON_SEARCH 查找包含特定元素的记录
SELECT name, preferences->'$.favorite_features' AS features 
FROM user_profiles 
WHERE JSON_SEARCH(preferences, 'one', 'analytics') IS NOT NULL;

-- 使用 JSON_SET 更新特定属性
UPDATE user_profiles 
SET preferences = JSON_SET(preferences, '$.notifications.push', true) 
WHERE id = %s;

高级 JSON 处理技巧

处理大型 JSON 数据

对于大型 JSON 数据，PyMySQL 提供了流式处理能力，可以通过 SSCursor（服务器端游标）来减少内存占用：

from pymysql.cursors import SSCursor

# 使用服务器端游标处理大型 JSON 数据
connection = pymysql.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    db='large_data_db',
    cursorclass=SSCursor
)

try:
    with connection.cursor() as cursor:
        # 处理包含大型 JSON 数据的表
        sql = "SELECT id, large_json_data FROM big_data_table"
        cursor.execute(sql)
        
        # 逐条处理记录，避免一次性加载所有数据
        for row in cursor:
            record_id, json_data = row
            process_large_json(record_id, json_data)
finally:
    connection.close()

自定义 JSON 编码器和解码器

PyMySQL 允许你自定义 JSON 数据的编码和解码逻辑，以满足特定需求。你可以通过修改转换器来自定义 JSON 处理行为：

import json
from datetime import datetime
from pymysql import converters

# 保存原始的 JSON 编码器
original_dict_encoder = converters.encoders[dict]

# 自定义 JSON 编码器，支持更多数据类型
def custom_json_encoder(value, mapping=None):
    if isinstance(value, dict):
        processed_dict = {}
        for k, v in value.items():
            if isinstance(v, datetime):
                processed_dict[k] = v.isoformat()
            else:
                processed_dict[k] = v
        json_str = json.dumps(processed_dict)
        return original_dict_encoder(json_str, mapping)
    return original_dict_encoder(value, mapping)

# 替换默认的字典编码器
converters.encoders[dict] = custom_json_encoder

批量操作 JSON 数据

PyMySQL 支持批量插入和更新 JSON 数据，提高处理效率：

try:
    with connection.cursor() as cursor:
        # 批量插入 JSON 数据
        insert_sql = "INSERT INTO user_profiles (name, preferences) VALUES (%s, %s)"
        
        users = [
            ("Alice", {"theme": "light", "notifications": {"email": True}}),
            ("Bob", {"theme": "dark", "notifications": {"email": False, "push": True}}),
            ("Charlie", {"theme": "blue", "favorite_features": ["dashboard", "reports"]})
        ]
        
        cursor.executemany(insert_sql, users)
        connection.commit()
        
        # 批量更新 JSON 字段
        update_sql = "UPDATE user_profiles SET preferences = JSON_SET(preferences, '$.notifications.sms', false)"
        cursor.execute(update_sql)
        connection.commit()
finally:
    connection.close()

常见问题与解决方案

JSON 解码错误

当 JSON 数据格式不正确或包含不受支持的数据类型时，PyMySQL 可能会抛出解码错误。解决方法是使用 try-except 捕获错误，并实现自定义错误处理：

try:
    cursor.execute("SELECT id, problematic_json FROM data_table")
    for row in cursor:
        try:
            json_data = row['problematic_json']
            # 处理 JSON 数据
        except (ValueError, TypeError) as e:
            print(f"Error parsing JSON for id {row['id']}: {e}")
            fix_corrupted_json(cursor, row['id'])
except pymysql.Error as e:
    print(f"MySQL error: {e}")

性能优化

处理大量 JSON 数据时，可以通过以下方法优化性能：

1. 创建 JSON 字段索引：对经常查询的 JSON 路径创建索引
2. 部分提取：只获取需要的 JSON 字段，减少数据传输
3. 批量操作：使用 executemany 减少网络往返
4. 连接池：使用连接池减少连接开销

-- 创建 JSON 字段索引示例
ALTER TABLE user_profiles ADD INDEX idx_preferences_theme ((preferences->'$.theme'));

总结

PyMySQL 提供了强大而灵活的 JSON 数据处理能力，通过自动类型转换和与 MySQL JSON 函数的无缝集成，使得 Python 与 MySQL 之间的 JSON 数据交互变得简单直观。无论是存储简单的配置信息，还是处理复杂的嵌套 JSON 结构，PyMySQL 都能满足你的需求。

通过本文介绍的方法，你可以：

• 使用 Python 字典和列表直接操作 MySQL JSON 数据
• 利用 MySQL JSON 函数进行高效的查询和更新
• 处理大型 JSON 数据和批量操作
• 自定义 JSON 编码和解码逻辑
• 优化 JSON 数据处理性能

要深入了解 PyMySQL 的 JSON 处理实现细节，可以查看 converters.py 源代码，该文件定义了所有数据类型的转换规则。更多 PyMySQL 使用示例可以参考官方文档。

掌握 PyMySQL 的 JSON 数据处理技巧，将帮助你在 Python 和 MySQL 开发中更灵活地应对复杂数据结构，构建更强大的应用程序。