From db0ab2c07716461e134af4ad6972d00bd93036fb Mon Sep 17 00:00:00 2001 From: junjun Date: Tue, 16 Sep 2025 14:43:30 +0800 Subject: [PATCH 1/5] feat: support Elasticsearch datasource #108 --- backend/apps/db/db.py | 27 +++++++++++++++------------ backend/apps/db/es_engine.py | 2 +- 2 files changed, 16 insertions(+), 13 deletions(-) diff --git a/backend/apps/db/db.py b/backend/apps/db/db.py index 41cf503a..495a88ff 100644 --- a/backend/apps/db/db.py +++ b/backend/apps/db/db.py @@ -420,15 +420,18 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= except Exception as ex: raise ParseSQLResultError(str(ex)) elif ds.type == 'es': - if table_name and table_name[0]: - res, columns = get_es_data(conf, sql, table_name[0]) - columns = [field[0] for field in columns] if origin_column else [field[0].lower() for - field in - columns] - result_list = [ - {str(columns[i]): float(value) if isinstance(value, Decimal) else value for i, value in - enumerate(tuple_item)} - for tuple_item in res - ] - return {"fields": columns, "data": result_list, - "sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))} + try: + if table_name and table_name[0]: + res, columns = get_es_data(conf, sql, table_name[0]) + columns = [field[0] for field in columns] if origin_column else [field[0].lower() for + field in + columns] + result_list = [ + {str(columns[i]): float(value) if isinstance(value, Decimal) else value for i, value in + enumerate(tuple_item)} + for tuple_item in res + ] + return {"fields": columns, "data": result_list, + "sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))} + except Exception as ex: + raise ParseSQLResultError(str(ex)) diff --git a/backend/apps/db/es_engine.py b/backend/apps/db/es_engine.py index 2373af99..6f7a1030 100644 --- a/backend/apps/db/es_engine.py +++ b/backend/apps/db/es_engine.py @@ -63,7 +63,7 @@ def get_es_fields(conf: DatasourceConf, table_name: str): def get_es_data(conf: DatasourceConf, sql: str, table_name: str): r = requests.post(f"{conf.host}/_sql/translate", json={"query": sql}) if r.json().get('error'): - raise HTTPException(status_code=500, detail=f': {json.dumps(r.json())}') + raise BaseException(f'{json.dumps(r.json())}') es_client = get_es_connect(conf) response = es_client.search( From 69bc20698cbffc5513bc498c6562f2dea6ac5a7d Mon Sep 17 00:00:00 2001 From: junjun Date: Tue, 16 Sep 2025 14:53:27 +0800 Subject: [PATCH 2/5] feat: support Elasticsearch datasource #108 --- backend/apps/db/db.py | 2 +- backend/apps/db/es_engine.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/apps/db/db.py b/backend/apps/db/db.py index 495a88ff..a1da1492 100644 --- a/backend/apps/db/db.py +++ b/backend/apps/db/db.py @@ -434,4 +434,4 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= return {"fields": columns, "data": result_list, "sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))} except Exception as ex: - raise ParseSQLResultError(str(ex)) + raise Exception(str(ex)) diff --git a/backend/apps/db/es_engine.py b/backend/apps/db/es_engine.py index 6f7a1030..05c5797b 100644 --- a/backend/apps/db/es_engine.py +++ b/backend/apps/db/es_engine.py @@ -63,7 +63,7 @@ def get_es_fields(conf: DatasourceConf, table_name: str): def get_es_data(conf: DatasourceConf, sql: str, table_name: str): r = requests.post(f"{conf.host}/_sql/translate", json={"query": sql}) if r.json().get('error'): - raise BaseException(f'{json.dumps(r.json())}') + print(json.dumps(r.json())) es_client = get_es_connect(conf) response = es_client.search( From 8dd4b1ee40d947c8d01c41d9a51afa3047d97363 Mon Sep 17 00:00:00 2001 From: junjun Date: Tue, 16 Sep 2025 15:14:03 +0800 Subject: [PATCH 3/5] feat: support Elasticsearch datasource #108 --- backend/apps/chat/task/llm.py | 1 + 1 file changed, 1 insertion(+) diff --git a/backend/apps/chat/task/llm.py b/backend/apps/chat/task/llm.py index 16736aa9..8cc1f470 100644 --- a/backend/apps/chat/task/llm.py +++ b/backend/apps/chat/task/llm.py @@ -1041,6 +1041,7 @@ class LLMService: real_execute_sql = assistant_dynamic_sql result = self.execute_sql(sql=real_execute_sql, tables=tables) + print(result) self.save_sql_data(data_obj=result) if in_chat: yield 'data:' + orjson.dumps({'content': 'execute-success', 'type': 'sql-data'}).decode() + '\n\n' From 7d2c972735b3d9018ed88dbc9d9adf4014e5f0d8 Mon Sep 17 00:00:00 2001 From: junjun Date: Tue, 16 Sep 2025 15:19:42 +0800 Subject: [PATCH 4/5] feat: support Elasticsearch datasource #108 --- backend/apps/chat/task/llm.py | 1 + backend/apps/db/db.py | 2 ++ 2 files changed, 3 insertions(+) diff --git a/backend/apps/chat/task/llm.py b/backend/apps/chat/task/llm.py index 8cc1f470..d8a422ea 100644 --- a/backend/apps/chat/task/llm.py +++ b/backend/apps/chat/task/llm.py @@ -1040,6 +1040,7 @@ class LLMService: subsql) real_execute_sql = assistant_dynamic_sql + print(tables) result = self.execute_sql(sql=real_execute_sql, tables=tables) print(result) self.save_sql_data(data_obj=result) diff --git a/backend/apps/db/db.py b/backend/apps/db/db.py index a1da1492..3a4dc94f 100644 --- a/backend/apps/db/db.py +++ b/backend/apps/db/db.py @@ -420,6 +420,7 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= except Exception as ex: raise ParseSQLResultError(str(ex)) elif ds.type == 'es': + print(table_name) try: if table_name and table_name[0]: res, columns = get_es_data(conf, sql, table_name[0]) @@ -431,6 +432,7 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= enumerate(tuple_item)} for tuple_item in res ] + print(result_list) return {"fields": columns, "data": result_list, "sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))} except Exception as ex: From 51e511a68a5cb3f0fdeff7c2e064649e9197c504 Mon Sep 17 00:00:00 2001 From: junjun Date: Tue, 16 Sep 2025 15:35:11 +0800 Subject: [PATCH 5/5] feat: support Elasticsearch datasource #108 --- backend/apps/chat/task/llm.py | 34 ++++++++++++++++------------------ backend/apps/db/db.py | 2 -- 2 files changed, 16 insertions(+), 20 deletions(-) diff --git a/backend/apps/chat/task/llm.py b/backend/apps/chat/task/llm.py index d8a422ea..30ac38a6 100644 --- a/backend/apps/chat/task/llm.py +++ b/backend/apps/chat/task/llm.py @@ -1000,26 +1000,26 @@ class LLMService: sqlbot_temp_sql_text = None assistant_dynamic_sql = None # todo row permission - if ((not self.current_assistant or is_page_embedded) and is_normal_user( - self.current_user)) or use_dynamic_ds: + if (not self.current_assistant or is_page_embedded) or use_dynamic_ds: sql, tables = self.check_sql(res=full_sql_text) sql_result = None - if use_dynamic_ds: - dynamic_sql_result = self.generate_assistant_dynamic_sql(sql, tables) - sqlbot_temp_sql_text = dynamic_sql_result.get( - 'sqlbot_temp_sql_text') if dynamic_sql_result else None - # sql_result = self.generate_assistant_filter(sql, tables) - else: - sql_result = self.generate_filter(sql, tables) # maybe no sql and tables + if is_normal_user(self.current_user): + if use_dynamic_ds: + dynamic_sql_result = self.generate_assistant_dynamic_sql(sql, tables) + sqlbot_temp_sql_text = dynamic_sql_result.get( + 'sqlbot_temp_sql_text') if dynamic_sql_result else None + # sql_result = self.generate_assistant_filter(sql, tables) + else: + sql_result = self.generate_filter(sql, tables) # maybe no sql and tables - if sql_result: - SQLBotLogUtil.info(sql_result) - sql = self.check_save_sql(res=sql_result) - elif dynamic_sql_result and sqlbot_temp_sql_text: - assistant_dynamic_sql = self.check_save_sql(res=sqlbot_temp_sql_text) - else: - sql = self.check_save_sql(res=full_sql_text) + if sql_result: + SQLBotLogUtil.info(sql_result) + sql = self.check_save_sql(res=sql_result) + elif dynamic_sql_result and sqlbot_temp_sql_text: + assistant_dynamic_sql = self.check_save_sql(res=sqlbot_temp_sql_text) + else: + sql = self.check_save_sql(res=full_sql_text) else: sql = self.check_save_sql(res=full_sql_text) tables = [] @@ -1040,9 +1040,7 @@ class LLMService: subsql) real_execute_sql = assistant_dynamic_sql - print(tables) result = self.execute_sql(sql=real_execute_sql, tables=tables) - print(result) self.save_sql_data(data_obj=result) if in_chat: yield 'data:' + orjson.dumps({'content': 'execute-success', 'type': 'sql-data'}).decode() + '\n\n' diff --git a/backend/apps/db/db.py b/backend/apps/db/db.py index 3a4dc94f..a1da1492 100644 --- a/backend/apps/db/db.py +++ b/backend/apps/db/db.py @@ -420,7 +420,6 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= except Exception as ex: raise ParseSQLResultError(str(ex)) elif ds.type == 'es': - print(table_name) try: if table_name and table_name[0]: res, columns = get_es_data(conf, sql, table_name[0]) @@ -432,7 +431,6 @@ def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column= enumerate(tuple_item)} for tuple_item in res ] - print(result_list) return {"fields": columns, "data": result_list, "sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))} except Exception as ex: