MySQL更新JSON字段key:value形式

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 = CASEWHEN JSON_SEARCH(props, 'one', :code, NULL, '$[*].code') IS NOT NULL THENJSON_SET(props,REPLACE(JSON_UNQUOTE(JSON_SEARCH(props, 'one', :code, NULL, '$[*].code')), '.code', '.value'),:result)ELSEJSON_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 = CASEWHEN JSON_SEARCH(props, 'one', 'test2', NULL, '$[*].code') IS NOT NULL THENJSON_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')ELSEJSON_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 部分的 SQLset_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"""ELSEJSON_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@staticmethoddef 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)# 格式化 SQLformatted_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})