当前位置：首页 > article >正文

MySQL更新JSON字段key:value形式

article 2024/12/1 4:59:36

MySQL更新JSON字段key:value形式

1. 介绍

‌MySQL的JSON数据类型‌是MySQL 5.7及以上版本中引入的一种数据类型，用于存储JSON格式的数据。使用JSON数据类型可以自动校验文档是否满足JSON格式的要求，优化存储格式，并允许快速访问文档中的特定元素，而无需读取整个文档

2. 针对key:value形式

初始场景：MySQL的JSON字段存储的数据形式

[
  {
    "code": "test",
    "value": "暂无"
  }

调用方法会获取到执行的value的值，依据code，如果存在进行更新，不存在进行添加。

update_sql = f"""
                        UPDATE {table_name}
                        SET props = CASE
                                      WHEN JSON_SEARCH(props, 'one', :code, NULL, '$[*].code') IS NOT NULL THEN
                                        JSON_SET(
                                            props,
                                            REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', :code, NULL, '$[*].code')), '.code', '.value'),
                                            :result
                                        )
                                      ELSE
                                        JSON_ARRAY_APPEND(
                                            IFNULL(props, JSON_ARRAY()), 
                                            '$', 
                                            JSON_OBJECT('code', :code, 'value', :result)
                                        )
                                    END,
                            updated_at = NOW()
                        WHERE id = :id
                    """

                    # 执行 SQL 更新
                    session_new.execute(text(update_sql), {
                        'code': code,
                        'result': result,
                        'id': id
                    })

进阶场景：MySQL的JSON字段存储的数据形式

[
  {
    "code": "test2",
    "value": "暂无",
    "update_time": "2024-11-28 19:13:12"
  }
]

获取的不再是单个value的值，而是一个dict,示例：

{
  "code": "test2",
  "value": "暂无数据",
  "test_id": 2,
  "all_test_id": 2,
  "aaaa": "12",
  "update_time": "2023-12-23 00:00:00"
}

依据code，如果存在进行更新，不存在进行添加，只是这次执行需要更新多个key:value形式数据。
示例，可执行sql:

UPDATE intention_extended_datas
SET 
    props = CASE
        WHEN JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code') IS NOT NULL THEN
            JSON_SET(
                JSON_SET(
                    JSON_SET(
                        JSON_SET(
                            JSON_SET(
                                props,
                                REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code')), '.code', '.value'),
                                '暂无'
                            ),
                            REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code')), '.code', '.test_id'),
                            2
                        ),
                        REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code')), '.code', '.all_test_id'),
                        3
                    ),
                    REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code')), '.code', '.aaa'),
                    '226'
                ),
                REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code')), '.code', '.update_time'),
                '2023-12-26 00:00:00'
            )
        ELSE
            JSON_ARRAY_APPEND(
                IFNULL(props, JSON_ARRAY()), 
                '$', 
                JSON_OBJECT(
                    'code', 'test2', 
                    'value', '暂无数据', 
                    'test_id', 1,
                    'all_test_id', 2, 
                    'aaa', '223',
                    'update_time', '2023-12-23 00:00:00'
                )
            )
    END,
    updated_at = NOW()
WHERE id = 1;

python代码如何实现这样的可执行sql,ressult是dict格式:

    def generate_update_sql(self, table_name, result) -> str:
        # 构建 SET 部分的 SQL
        set_sql_parts = []

        # 构建 WHEN 子句
        when_clause = f"""
            WHEN JSON_SEARCH(props, 'one', :code, NULL, '$[*].code') IS NOT NULL THEN
                {self.construct_json_set_query(result)}
        """
        set_sql_parts.append(when_clause)

        # 构建 ELSE 子句，添加新的元素到 JSON 数组
        else_clause = f"""
            ELSE
                JSON_ARRAY_APPEND(
                    IFNULL(props, JSON_ARRAY()),
                    '$',
                    JSON_OBJECT(
                        'code', :code,
                        {', '.join([f"'{key}', :{key}" for key in result.keys()])}
                    )
                )
        """
        set_sql_parts.append(else_clause)

        # 构建更新语句
        set_sql = f"""
            UPDATE {table_name}
            SET props = CASE
                {" ".join(set_sql_parts)}
            END,
            updated_at = NOW()
            WHERE id = :id;
        """

        return set_sql

    @staticmethod
    def construct_json_set_query(dynamic_data: dict) -> str:
        """
        根据传入的动态字典，构造 JSON_SET 的嵌套语句。

        :param dynamic_data: 动态字典，键为路径后缀，值为占位符。
                             示例：{"value": ":value", "test_id": ":test_id"}
        :return: 返回生成的 SQL JSON_SET 嵌套语句。
        """
        if not dynamic_data:
            return "props"  # 如果没有键值对，直接返回基础结构

        base_path = "REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', :code, NULL, '$[*].code')), '.code', '{}')"
        json_set_template = "JSON_SET({}, {}, {})"

        # 构造嵌套 JSON_SET 语句
        nested_set = "props"
        for key, placeholder in dynamic_data.items():
            replace_path = base_path.format(f".{key}")  # 替换路径
            nested_set = json_set_template.format(nested_set, replace_path, f':{key}')

        return nested_set

外部调用：

# 如果是 JSON 字段，需要检查是否存在，并进行更新或追加
                    update_sql = self.generate_update_sql(table_name, result)
                    # 格式化 SQL
                    formatted_sql = sqlparse.format(update_sql, reindent=True, keyword_case='upper')
                    # 执行 SQL 更新，将 result 中的所有键值对与 code 和 id 一起传递
                    session_new.execute(text(formatted_sql), {
                        'code': code,
                        'id': id,
                        **result
                    })