MySQL更新JSON字段key:value形式

1. 介紹

?MySQL的JSON數(shù)據(jù)類型?是MySQL 5.7及以上版本中引入的一種數(shù)據(jù)類型，用于存儲JSON格式的數(shù)據(jù)。使用JSON數(shù)據(jù)類型可以自動校驗文檔是否滿足JSON格式的要求，優(yōu)化存儲格式，并允許快速訪問文檔中的特定元素，而無需讀取整個文檔

2. 針對key:value形式

初始場景：MySQL的JSON字段存儲的數(shù)據(jù)形式

[{"code": "test","value": "暫無"}

調(diào)用方法會獲取到執(zhí)行的value的值，依據(jù)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"""# 執(zhí)行 SQL 更新session_new.execute(text(update_sql), {'code': code,'result': result,'id': id})

進階場景：MySQL的JSON字段存儲的數(shù)據(jù)形式

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

獲取的不再是單個value的值，而是一個dict,示例：

{"code": "test2","value": "暫無數(shù)據(jù)","test_id": 2,"all_test_id": 2,"aaaa": "12","update_time": "2023-12-23 00:00:00"
}

依據(jù)code，如果存在進行更新，不存在進行添加，只是這次執(zhí)行需要更新多個key:value形式數(shù)據(jù)。
示例，可執(zhí)行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', '暫無數(shù)據(jù)', '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代碼如何實現(xiàn)這樣的可執(zhí)行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 數(shù)組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:"""根據(jù)傳入的動態(tài)字典，構造 JSON_SET 的嵌套語句。:param dynamic_data: 動態(tài)字典，鍵為路徑后綴，值為占位符。示例：{"value": ":value", "test_id": ":test_id"}:return: 返回生成的 SQL JSON_SET 嵌套語句。"""if not dynamic_data:return "props"  # 如果沒有鍵值對，直接返回基礎結(jié)構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

外部調(diào)用：

# 如果是 JSON 字段，需要檢查是否存在，并進行更新或追加update_sql = self.generate_update_sql(table_name, result)# 格式化 SQLformatted_sql = sqlparse.format(update_sql, reindent=True, keyword_case='upper')# 執(zhí)行 SQL 更新，將 result 中的所有鍵值對與 code 和 id 一起傳遞session_new.execute(text(formatted_sql), {'code': code,'id': id,**result})