diff --git a/sql/data_dictionary.py b/sql/data_dictionary.py
index 6e5bd19cdc..401e17c8df 100644
--- a/sql/data_dictionary.py
+++ b/sql/data_dictionary.py
@@ -89,6 +89,176 @@ def table_info(request):
)
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def view_list(request):
+ """数据字典获取视图列表（仅MySQL）"""
+ return _dict_list(request, db_type_required="mysql", engine_method="get_views_list")
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def view_info(request):
+ """数据字典获取视图详情（仅MySQL）"""
+ return _dict_detail(
+ request,
+ db_type_required="mysql",
+ engine_method="get_view_detail",
+ name_param="view_name",
+ engine_kwarg="view_name",
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def trigger_list(request):
+ """数据字典获取触发器列表（仅MySQL）"""
+ return _dict_list(
+ request, db_type_required="mysql", engine_method="get_triggers_list"
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def trigger_info(request):
+ """数据字典获取触发器详情（仅MySQL）"""
+ return _dict_detail(
+ request,
+ db_type_required="mysql",
+ engine_method="get_trigger_detail",
+ name_param="trigger_name",
+ engine_kwarg="trigger_name",
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def procedure_list(request):
+ """数据字典获取存储过程列表（仅MySQL）"""
+ return _dict_list(
+ request, db_type_required="mysql", engine_method="get_procedures_list"
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def procedure_info(request):
+ """数据字典获取存储过程详情（仅MySQL）"""
+ return _dict_detail(
+ request,
+ db_type_required="mysql",
+ engine_method="get_procedure_detail",
+ name_param="proc_name",
+ engine_kwarg="proc_name",
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def function_list(request):
+ """数据字典获取函数列表（仅MySQL）"""
+ return _dict_list(
+ request, db_type_required="mysql", engine_method="get_functions_list"
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def function_info(request):
+ """数据字典获取函数详情（仅MySQL）"""
+ return _dict_detail(
+ request,
+ db_type_required="mysql",
+ engine_method="get_function_detail",
+ name_param="func_name",
+ engine_kwarg="func_name",
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def event_list(request):
+ """数据字典获取定时任务列表（仅MySQL）"""
+ return _dict_list(
+ request, db_type_required="mysql", engine_method="get_events_list"
+ )
+
+
+@permission_required("sql.menu_data_dictionary", raise_exception=True)
+def event_info(request):
+ """数据字典获取定时任务详情（仅MySQL）"""
+ return _dict_detail(
+ request,
+ db_type_required="mysql",
+ engine_method="get_event_detail",
+ name_param="event_name",
+ engine_kwarg="event_name",
+ )
+
+
+def _dict_list(request, db_type_required, engine_method):
+ """通用数据字典对象列表接口"""
+ instance_name = request.GET.get("instance_name", "")
+ db_name = request.GET.get("db_name", "")
+ db_type = request.GET.get("db_type", "")
+
+ if db_type_required and db_type != db_type_required:
+ res = {"status": 1, "msg": "仅MySQL支持该功能"}
+ return HttpResponse(
+ json.dumps(res, cls=ExtendJSONEncoder, bigint_as_string=True),
+ content_type="application/json",
+ )
+
+ if instance_name and db_name:
+ try:
+ instance = Instance.objects.get(
+ instance_name=instance_name, db_type=db_type
+ )
+ query_engine = get_engine(instance=instance)
+ db_name = query_engine.escape_string(db_name)
+ data = getattr(query_engine, engine_method)(db_name=db_name)
+ res = {"status": 0, "data": data}
+ except Instance.DoesNotExist:
+ res = {"status": 1, "msg": "Instance.DoesNotExist"}
+ except Exception as e:
+ res = {"status": 1, "msg": str(e)}
+ else:
+ res = {"status": 1, "msg": "非法调用！"}
+ return HttpResponse(
+ json.dumps(res, cls=ExtendJSONEncoder, bigint_as_string=True),
+ content_type="application/json",
+ )
+
+
+def _dict_detail(request, db_type_required, engine_method, name_param, engine_kwarg):
+ """通用数据字典对象详情接口"""
+ instance_name = request.GET.get("instance_name", "")
+ db_name = request.GET.get("db_name", "")
+ obj_name = request.GET.get(name_param, "")
+ db_type = request.GET.get("db_type", "")
+
+ if db_type_required and db_type != db_type_required:
+ res = {"status": 1, "msg": "仅MySQL支持该功能"}
+ return HttpResponse(
+ json.dumps(res, cls=ExtendJSONEncoder, bigint_as_string=True),
+ content_type="application/json",
+ )
+
+ if instance_name and db_name and obj_name:
+ try:
+ instance = Instance.objects.get(
+ instance_name=instance_name, db_type=db_type
+ )
+ query_engine = get_engine(instance=instance)
+ db_name = query_engine.escape_string(db_name)
+ obj_name = query_engine.escape_string(obj_name)
+ data = getattr(query_engine, engine_method)(
+ **{"db_name": db_name, engine_kwarg: obj_name}
+ )
+ res = {"status": 0, "data": data}
+ except Instance.DoesNotExist:
+ res = {"status": 1, "msg": "Instance.DoesNotExist"}
+ except Exception as e:
+ res = {"status": 1, "msg": str(e)}
+ else:
+ res = {"status": 1, "msg": "非法调用！"}
+ return HttpResponse(
+ json.dumps(res, cls=ExtendJSONEncoder, bigint_as_string=True),
+ content_type="application/json",
+ )
+
+
def get_export_full_path(base_dir: str, instance_name: str, db_name: str) -> str:
"""validate if the instance_name and db_name provided is secure"""
fullpath = os.path.normpath(
diff --git a/sql/engines/__init__.py b/sql/engines/__init__.py
index 27f31b3c14..d2083ac049 100644
--- a/sql/engines/__init__.py
+++ b/sql/engines/__init__.py
@@ -139,6 +139,46 @@ def get_tables_metas_data(self, db_name, **kwargs):
"""获取数据库所有表格信息，用作数据字典导出接口"""
return list()
+ def get_views_list(self, db_name, **kwargs):
+ """获取视图列表, 返回 dict"""
+ return dict()
+
+ def get_view_detail(self, db_name, view_name, **kwargs):
+ """获取视图详情, 返回 dict"""
+ return dict()
+
+ def get_triggers_list(self, db_name, **kwargs):
+ """获取触发器列表, 返回 dict"""
+ return dict()
+
+ def get_trigger_detail(self, db_name, trigger_name, **kwargs):
+ """获取触发器详情, 返回 dict"""
+ return dict()
+
+ def get_procedures_list(self, db_name, **kwargs):
+ """获取存储过程列表, 返回 dict"""
+ return dict()
+
+ def get_procedure_detail(self, db_name, proc_name, **kwargs):
+ """获取存储过程详情, 返回 dict"""
+ return dict()
+
+ def get_functions_list(self, db_name, **kwargs):
+ """获取函数列表, 返回 dict"""
+ return dict()
+
+ def get_function_detail(self, db_name, func_name, **kwargs):
+ """获取函数详情, 返回 dict"""
+ return dict()
+
+ def get_events_list(self, db_name, **kwargs):
+ """获取定时任务列表, 返回 dict"""
+ return dict()
+
+ def get_event_detail(self, db_name, event_name, **kwargs):
+ """获取定时任务详情, 返回 dict"""
+ return dict()
+
def get_all_databases_summary(self):
"""实例数据库管理功能，获取实例所有的数据库描述信息"""
return ResultSet()
diff --git a/sql/engines/mysql.py b/sql/engines/mysql.py
index d94c711ddc..f8a8da1ede 100644
--- a/sql/engines/mysql.py
+++ b/sql/engines/mysql.py
@@ -299,6 +299,244 @@ def get_table_index_data(self, db_name, tb_name, **kwargs):
)
return {"column_list": _index_data.column_list, "rows": _index_data.rows}
+ def get_views_list(self, db_name, **kwargs):
+ """获取视图列表，按首字符分组"""
+ data = {}
+ sql = """SELECT TABLE_NAME, VIEW_DEFINITION
+ FROM information_schema.VIEWS
+ WHERE TABLE_SCHEMA=%(db_name)s;"""
+ result = self.query(db_name=db_name, sql=sql, parameters={"db_name": db_name})
+ for row in result.rows:
+ view_name = row[0]
+ view_comment = row[1][:80] if row[1] else ""
+ if view_name[0] not in data:
+ data[view_name[0]] = list()
+ data[view_name[0]].append([view_name, view_comment])
+ return data
+
+ def get_view_detail(self, db_name, view_name, **kwargs):
+ """获取视图详情"""
+ sql = """SELECT
+ TABLE_NAME as view_name,
+ VIEW_DEFINITION as view_definition,
+ CHECK_OPTION as check_option,
+ IS_UPDATABLE as is_updatable,
+ DEFINER as definer,
+ SECURITY_TYPE as security_type,
+ CHARACTER_SET_CLIENT as character_set_client,
+ COLLATION_CONNECTION as collation_connection
+ FROM information_schema.VIEWS
+ WHERE TABLE_SCHEMA=%(db_name)s AND TABLE_NAME=%(view_name)s;"""
+ _meta = self.query(
+ db_name, sql, parameters={"db_name": db_name, "view_name": view_name}
+ )
+ meta_data = {
+ "column_list": _meta.column_list,
+ "rows": _meta.rows[0] if _meta.rows else [],
+ }
+ view_definition = ""
+ if _meta.rows:
+ # VIEW_DEFINITION 在第二列
+ view_definition = _meta.rows[0][1] or ""
+ desc = self.get_table_desc_data(db_name=db_name, tb_name=view_name)
+ return {
+ "meta_data": meta_data,
+ "desc": desc,
+ "view_definition": view_definition,
+ }
+
+ def get_triggers_list(self, db_name, **kwargs):
+ """获取触发器列表，按首字符分组"""
+ data = {}
+ sql = """SELECT
+ TRIGGER_NAME,
+ ACTION_TIMING,
+ EVENT_MANIPULATION,
+ EVENT_OBJECT_TABLE
+ FROM information_schema.TRIGGERS
+ WHERE TRIGGER_SCHEMA=%(db_name)s;"""
+ result = self.query(db_name=db_name, sql=sql, parameters={"db_name": db_name})
+ for row in result.rows:
+ trigger_name = row[0]
+ desc = f"{row[1]} {row[2]} ON {row[3]}"
+ if trigger_name[0] not in data:
+ data[trigger_name[0]] = list()
+ data[trigger_name[0]].append([trigger_name, desc])
+ return data
+
+ def get_trigger_detail(self, db_name, trigger_name, **kwargs):
+ """获取触发器详情"""
+ sql = """SELECT
+ TRIGGER_NAME as trigger_name,
+ ACTION_TIMING as action_timing,
+ EVENT_MANIPULATION as event_manipulation,
+ EVENT_OBJECT_TABLE as event_object_table,
+ ACTION_ORIENTATION as action_orientation,
+ ACTION_STATEMENT as action_statement,
+ DEFINER as definer,
+ CREATED as created,
+ SQL_MODE as sql_mode,
+ CHARACTER_SET_CLIENT as character_set_client,
+ COLLATION_CONNECTION as collation_connection
+ FROM information_schema.TRIGGERS
+ WHERE TRIGGER_SCHEMA=%(db_name)s AND TRIGGER_NAME=%(trigger_name)s;"""
+ _data = self.query(
+ db_name,
+ sql,
+ parameters={"db_name": db_name, "trigger_name": trigger_name},
+ )
+ return {
+ "column_list": _data.column_list,
+ "rows": _data.rows[0] if _data.rows else [],
+ }
+
+ def get_procedures_list(self, db_name, **kwargs):
+ """获取存储过程列表，按首字符分组"""
+ data = {}
+ sql = """SELECT ROUTINE_NAME, ROUTINE_COMMENT
+ FROM information_schema.ROUTINES
+ WHERE ROUTINE_SCHEMA=%(db_name)s AND ROUTINE_TYPE='PROCEDURE';"""
+ result = self.query(db_name=db_name, sql=sql, parameters={"db_name": db_name})
+ for row in result.rows:
+ proc_name = row[0]
+ proc_cmt = row[1]
+ if proc_name[0] not in data:
+ data[proc_name[0]] = list()
+ data[proc_name[0]].append([proc_name, proc_cmt])
+ return data
+
+ def get_procedure_detail(self, db_name, proc_name, **kwargs):
+ """获取存储过程详情"""
+ sql_meta = """SELECT
+ ROUTINE_NAME as routine_name,
+ ROUTINE_SCHEMA as routine_schema,
+ DEFINER as definer,
+ CREATED as created,
+ LAST_ALTERED as last_altered,
+ SQL_MODE as sql_mode,
+ SECURITY_TYPE as security_type,
+ ROUTINE_COMMENT as routine_comment
+ FROM information_schema.ROUTINES
+ WHERE ROUTINE_SCHEMA=%(db_name)s
+ AND ROUTINE_NAME=%(proc_name)s
+ AND ROUTINE_TYPE='PROCEDURE';"""
+ _meta = self.query(
+ db_name,
+ sql_meta,
+ parameters={"db_name": db_name, "proc_name": proc_name},
+ )
+ meta_data = {
+ "column_list": _meta.column_list,
+ "rows": _meta.rows[0] if _meta.rows else [],
+ }
+ _create = self.query(db_name, f"SHOW CREATE PROCEDURE `{proc_name}`;")
+ create_sql = _create.rows
+ return {"meta_data": meta_data, "create_sql": create_sql}
+
+ def get_functions_list(self, db_name, **kwargs):
+ """获取函数列表，按首字符分组"""
+ data = {}
+ sql = """SELECT ROUTINE_NAME, ROUTINE_COMMENT
+ FROM information_schema.ROUTINES
+ WHERE ROUTINE_SCHEMA=%(db_name)s AND ROUTINE_TYPE='FUNCTION';"""
+ result = self.query(db_name=db_name, sql=sql, parameters={"db_name": db_name})
+ for row in result.rows:
+ func_name = row[0]
+ func_cmt = row[1]
+ if func_name[0] not in data:
+ data[func_name[0]] = list()
+ data[func_name[0]].append([func_name, func_cmt])
+ return data
+
+ def get_function_detail(self, db_name, func_name, **kwargs):
+ """获取函数详情"""
+ sql_meta = """SELECT
+ ROUTINE_NAME as routine_name,
+ ROUTINE_SCHEMA as routine_schema,
+ DTD_IDENTIFIER as return_type,
+ DEFINER as definer,
+ CREATED as created,
+ LAST_ALTERED as last_altered,
+ SQL_MODE as sql_mode,
+ SECURITY_TYPE as security_type,
+ ROUTINE_COMMENT as routine_comment
+ FROM information_schema.ROUTINES
+ WHERE ROUTINE_SCHEMA=%(db_name)s
+ AND ROUTINE_NAME=%(func_name)s
+ AND ROUTINE_TYPE='FUNCTION';"""
+ _meta = self.query(
+ db_name,
+ sql_meta,
+ parameters={"db_name": db_name, "func_name": func_name},
+ )
+ meta_data = {
+ "column_list": _meta.column_list,
+ "rows": _meta.rows[0] if _meta.rows else [],
+ }
+ _create = self.query(db_name, f"SHOW CREATE FUNCTION `{func_name}`;")
+ create_sql = _create.rows
+ return {"meta_data": meta_data, "create_sql": create_sql}
+
+ def get_events_list(self, db_name, **kwargs):
+ """获取定时任务列表，按首字符分组"""
+ data = {}
+ sql = """SELECT
+ EVENT_NAME,
+ STATUS,
+ EVENT_TYPE,
+ INTERVAL_VALUE,
+ INTERVAL_FIELD
+ FROM information_schema.EVENTS
+ WHERE EVENT_SCHEMA=%(db_name)s;"""
+ result = self.query(db_name=db_name, sql=sql, parameters={"db_name": db_name})
+ for row in result.rows:
+ event_name = row[0]
+ status = row[1]
+ event_type = row[2]
+ interval_value = row[3]
+ interval_field = row[4]
+ if event_type == "RECURRING":
+ desc = f"{status} EVERY {interval_value} {interval_field}"
+ else:
+ desc = f"{status} ONE TIME"
+ if event_name[0] not in data:
+ data[event_name[0]] = list()
+ data[event_name[0]].append([event_name, desc])
+ return data
+
+ def get_event_detail(self, db_name, event_name, **kwargs):
+ """获取定时任务详情"""
+ sql_meta = """SELECT
+ EVENT_NAME as event_name,
+ EVENT_SCHEMA as event_schema,
+ DEFINER as definer,
+ EVENT_TYPE as event_type,
+ INTERVAL_VALUE as interval_value,
+ INTERVAL_FIELD as interval_field,
+ STATUS as status,
+ EXECUTE_AT as execute_at,
+ STARTS as starts,
+ ENDS as ends,
+ LAST_EXECUTED as last_executed,
+ ON_COMPLETION as on_completion,
+ CREATED as created,
+ LAST_ALTERED as last_altered,
+ EVENT_COMMENT as event_comment
+ FROM information_schema.EVENTS
+ WHERE EVENT_SCHEMA=%(db_name)s AND EVENT_NAME=%(event_name)s;"""
+ _meta = self.query(
+ db_name,
+ sql_meta,
+ parameters={"db_name": db_name, "event_name": event_name},
+ )
+ meta_data = {
+ "column_list": _meta.column_list,
+ "rows": _meta.rows[0] if _meta.rows else [],
+ }
+ _create = self.query(db_name, f"SHOW CREATE EVENT `{event_name}`;")
+ create_sql = _create.rows
+ return {"meta_data": meta_data, "create_sql": create_sql}
+
def get_tables_metas_data(self, db_name, **kwargs):
"""获取数据库所有表格信息，用作数据字典导出接口"""
sql_tbs = f"SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=%(db_name)s ORDER BY TABLE_SCHEMA,TABLE_NAME;"
diff --git a/sql/engines/test_mysql.py b/sql/engines/test_mysql.py
index 9aa34506e3..4fd0b0f7de 100644
--- a/sql/engines/test_mysql.py
+++ b/sql/engines/test_mysql.py
@@ -188,6 +188,230 @@ def testDescribe(self, mock_query):
new_engine.describe_table("some_db", "some_db")
mock_query.assert_called_once()
+ @patch.object(MysqlEngine, "query")
+ def test_get_views_list(self, mock_query):
+ r = ResultSet()
+ r.rows = [("v1", "select 1"), ("v2", "select 2"), ("a_view", None)]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_views_list(db_name="some_db")
+ self.assertIn("v", data)
+ self.assertIn("a", data)
+ self.assertEqual(data["v"][0][0], "v1")
+
+ @patch.object(MysqlEngine, "get_table_desc_data")
+ @patch.object(MysqlEngine, "query")
+ def test_get_view_detail(self, mock_query, mock_desc):
+ r = ResultSet()
+ r.column_list = [
+ "view_name",
+ "view_definition",
+ "check_option",
+ "is_updatable",
+ "definer",
+ "security_type",
+ "character_set_client",
+ "collation_connection",
+ ]
+ r.rows = [
+ (
+ "v1",
+ "select 1",
+ "NONE",
+ "YES",
+ "root@%",
+ "DEFINER",
+ "utf8",
+ "utf8_general_ci",
+ )
+ ]
+ mock_query.return_value = r
+ mock_desc.return_value = {"column_list": [], "rows": []}
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_view_detail(db_name="some_db", view_name="v1")
+ self.assertEqual(data["view_definition"], "select 1")
+ self.assertIn("meta_data", data)
+ self.assertIn("desc", data)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_triggers_list(self, mock_query):
+ r = ResultSet()
+ r.rows = [("tg1", "BEFORE", "INSERT", "t1")]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_triggers_list(db_name="some_db")
+ self.assertIn("t", data)
+ self.assertEqual(data["t"][0][0], "tg1")
+ self.assertIn("BEFORE", data["t"][0][1])
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_trigger_detail(self, mock_query):
+ r = ResultSet()
+ r.column_list = [
+ "trigger_name",
+ "action_timing",
+ "event_manipulation",
+ "event_object_table",
+ "action_orientation",
+ "action_statement",
+ "definer",
+ "created",
+ "sql_mode",
+ "character_set_client",
+ "collation_connection",
+ ]
+ r.rows = [
+ (
+ "tg1",
+ "BEFORE",
+ "INSERT",
+ "t1",
+ "ROW",
+ "BEGIN END",
+ "root@%",
+ None,
+ "",
+ "utf8",
+ "utf8",
+ )
+ ]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_trigger_detail(db_name="some_db", trigger_name="tg1")
+ self.assertIn("column_list", data)
+ self.assertTrue(len(data["rows"]) > 0)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_procedures_list(self, mock_query):
+ r = ResultSet()
+ r.rows = [("p1", "comment1"), ("p2", "comment2")]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_procedures_list(db_name="some_db")
+ self.assertIn("p", data)
+ self.assertEqual(len(data["p"]), 2)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_procedure_detail(self, mock_query):
+ meta = ResultSet()
+ meta.column_list = [
+ "routine_name",
+ "routine_schema",
+ "definer",
+ "created",
+ "last_altered",
+ "sql_mode",
+ "security_type",
+ "routine_comment",
+ ]
+ meta.rows = [("p1", "some_db", "root@%", None, None, "", "DEFINER", "")]
+ create = ResultSet()
+ create.rows = [("p1", "", "CREATE PROCEDURE p1() BEGIN END", "")]
+ mock_query.side_effect = [meta, create]
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_procedure_detail(db_name="some_db", proc_name="p1")
+ self.assertIn("meta_data", data)
+ self.assertEqual(data["create_sql"], create.rows)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_functions_list(self, mock_query):
+ r = ResultSet()
+ r.rows = [("f1", "cmt"), ("f2", "cmt2")]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_functions_list(db_name="some_db")
+ self.assertIn("f", data)
+ self.assertEqual(len(data["f"]), 2)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_function_detail(self, mock_query):
+ meta = ResultSet()
+ meta.column_list = [
+ "routine_name",
+ "routine_schema",
+ "return_type",
+ "definer",
+ "created",
+ "last_altered",
+ "sql_mode",
+ "security_type",
+ "routine_comment",
+ ]
+ meta.rows = [("f1", "some_db", "int", "root@%", None, None, "", "DEFINER", "")]
+ create = ResultSet()
+ create.rows = [("f1", "", "CREATE FUNCTION f1() RETURNS INT RETURN 1", "", "")]
+ mock_query.side_effect = [meta, create]
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_function_detail(db_name="some_db", func_name="f1")
+ self.assertIn("meta_data", data)
+ self.assertEqual(data["create_sql"], create.rows)
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_events_list(self, mock_query):
+ r = ResultSet()
+ r.rows = [("e1", "ENABLED", "RECURRING", 1, "DAY")]
+ mock_query.return_value = r
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_events_list(db_name="some_db")
+ self.assertIn("e", data)
+ self.assertIn("EVERY", data["e"][0][1])
+
+ @patch.object(MysqlEngine, "query")
+ def test_get_event_detail(self, mock_query):
+ meta = ResultSet()
+ meta.column_list = [
+ "event_name",
+ "event_schema",
+ "definer",
+ "event_type",
+ "interval_value",
+ "interval_field",
+ "status",
+ "execute_at",
+ "starts",
+ "ends",
+ "last_executed",
+ "on_completion",
+ "created",
+ "last_altered",
+ "event_comment",
+ ]
+ meta.rows = [
+ (
+ "e1",
+ "some_db",
+ "root@%",
+ "RECURRING",
+ 1,
+ "DAY",
+ "ENABLED",
+ None,
+ None,
+ None,
+ None,
+ "NOT PRESERVE",
+ None,
+ None,
+ "",
+ )
+ ]
+ create = ResultSet()
+ create.rows = [
+ (
+ "e1",
+ "",
+ "",
+ "CREATE EVENT e1 ON SCHEDULE EVERY 1 DAY DO SELECT 1",
+ "",
+ "",
+ )
+ ]
+ mock_query.side_effect = [meta, create]
+ engine = MysqlEngine(instance=self.ins1)
+ data = engine.get_event_detail(db_name="some_db", event_name="e1")
+ self.assertIn("meta_data", data)
+ self.assertEqual(data["create_sql"], create.rows)
+
def testQueryCheck(self):
new_engine = MysqlEngine(instance=self.ins1)
sql_without_limit = "-- 测试\n select user from usertable"
diff --git a/sql/templates/data_dictionary.html b/sql/templates/data_dictionary.html
index 238d9b4875..ea9f22ffff 100644
--- a/sql/templates/data_dictionary.html
+++ b/sql/templates/data_dictionary.html
@@ -23,6 +23,18 @@
data-live-search="true">
+
+
+
{% if perms.sql.data_dictionary_export %}
+
+
+
+
+
@@ -410,6 +662,24 @@ 表元数据展示
//实例变动获取库
$("#instance_name").change(function () {
$('#db_name').empty();
+ // 根据 db_type 调整 obj_type 可选项
+ var _ins_name = $("#instance_name").val();
+ var _ind = instance_result.findIndex((value)=>value.instance_name==_ins_name);
+ var _db_type = _ind >= 0 ? instance_result[_ind]['db_type'] : '';
+ if (_db_type === 'mysql') {
+ $('#obj_type option').prop('disabled', false);
+ } else {
+ // 非MySQL仅显示“表”
+ $('#obj_type option').each(function () {
+ if ($(this).val() === 'table') {
+ $(this).prop('disabled', false);
+ } else {
+ $(this).prop('disabled', true);
+ }
+ });
+ $('#obj_type').val('table');
+ }
+ $('#obj_type').selectpicker('refresh');
$.ajax({
type: "get",
url: "/instance/instance_resource/",
@@ -446,9 +716,40 @@ 表元数据展示
//库变动获取表
$("#db_name").change(function () {
- get_table_list()
+ getObjectList();
});
+ //对象类型变动重新加载
+ $("#obj_type").change(function () {
+ getObjectList();
+ });
+
+ // 获取当前选择的实例 db_type
+ function getCurrentDbType() {
+ var instance_name = $("#instance_name").val();
+ if (!instance_name) return '';
+ var ind = instance_result.findIndex((value)=>value.instance_name==instance_name);
+ if (ind < 0) return '';
+ return instance_result[ind]['db_type'];
+ }
+
+ // 统一对象列表入口
+ function getObjectList() {
+ var obj_type = $('#obj_type').val() || 'table';
+ var db_type = getCurrentDbType();
+ // 非MySQL实例强制重置为 table
+ if (db_type && db_type !== 'mysql' && obj_type !== 'table') {
+ $('#obj_type').val('table');
+ $('#obj_type').selectpicker('refresh');
+ obj_type = 'table';
+ }
+ if (obj_type === 'table') {
+ get_table_list();
+ } else {
+ get_dict_list(obj_type);
+ }
+ }
+
// 获取表
function get_table_list() {
var instance_name = $("#instance_name").val();
@@ -645,5 +946,361 @@ 表元数据展示
}
});
}
+
+ // 对象类型到接口、详情函数的映射
+ var OBJ_TYPE_CONFIG = {
+ view: {
+ listUrl: '/data_dictionary/view_list/',
+ showFn: 'showViewInfo'
+ },
+ trigger: {
+ listUrl: '/data_dictionary/trigger_list/',
+ showFn: 'showTriggerInfo'
+ },
+ procedure: {
+ listUrl: '/data_dictionary/procedure_list/',
+ showFn: 'showProcedureInfo'
+ },
+ function: {
+ listUrl: '/data_dictionary/function_list/',
+ showFn: 'showFunctionInfo'
+ },
+ event: {
+ listUrl: '/data_dictionary/event_list/',
+ showFn: 'showEventInfo'
+ }
+ };
+
+ // 通用列表获取（视图/触发器/存储过程/函数/事件）
+ function get_dict_list(obj_type) {
+ var conf = OBJ_TYPE_CONFIG[obj_type];
+ if (!conf) return;
+ var instance_name = $("#instance_name").val();
+ var db_name = $('#db_name').val();
+ var db_type = getCurrentDbType();
+ if (instance_name === "" || db_name === "") {
+ alert("请先选择实例和数据库！");
+ return;
+ }
+ $.ajax({
+ type: "get",
+ url: conf.listUrl,
+ dataType: "json",
+ data: {
+ instance_name: instance_name,
+ db_name: db_name,
+ db_type: db_type
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ $('#btn_export_dict').removeClass('disabled');
+ $('#btn_export_dict').prop('disabled', false);
+ $('#jumpbox').empty();
+ $('#indexTable').empty();
+ var result = data.data;
+ for (var k in result) {
+ var jumpBoxStr = '' + k + '';
+ $('#jumpbox').append(jumpBoxStr);
+ $('#indexTable').append('' +
+ ' | ' +
+ '' + k + ' | ' +
+ ' | ' +
+ '
');
+ for (var i = 0; i < result[k].length; i++) {
+ var objName = result[k][i][0];
+ var objDesc = result[k][i][1] == null ? '' : result[k][i][1];
+ var indexTableStr = '' +
+ ' | ' +
+ '' +
+ '' + objName + '' +
+ ' | ' +
+ '' + objDesc + ' | ' +
+ '
';
+ $('#indexTable').append(indexTableStr);
+ }
+ }
+ $('#indexTable').prepend('| |
');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
+
+ // 填充 meta_data table，通用工具
+ function fillMetaTable(tableSelector, idPrefix, meta_data) {
+ if (!meta_data || !meta_data.column_list) return;
+ var cols = meta_data.column_list;
+ var vals = meta_data.rows || [];
+ for (var i = 0; i < cols.length; i++) {
+ var v = vals[i];
+ if (v === null || v === undefined) v = '';
+ $(tableSelector + ' #' + idPrefix + cols[i]).text(v);
+ }
+ }
+
+ // 格式化 SQL 到 pre 元素
+ function renderSqlToPre(preSelector, sqlText) {
+ if (sqlText === null || sqlText === undefined) sqlText = '';
+ try {
+ if (window.sqlFormatter && typeof window.sqlFormatter.format === 'function') {
+ sqlText = window.sqlFormatter.format(sqlText);
+ }
+ } catch (e) { /* ignore */ }
+ $(preSelector).text(sqlText);
+ }
+
+ // 展示视图详情
+ function showViewInfo(ins_name, db_name, view_name) {
+ $.ajax({
+ type: "get",
+ url: "/data_dictionary/view_info/",
+ dataType: "json",
+ data: {
+ instance_name: ins_name,
+ db_name: db_name,
+ view_name: view_name,
+ db_type: 'mysql'
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ var result = data.data;
+ showModalBody('view');
+ // 填充基本信息
+ var meta = result['meta_data'] || {};
+ var cols = meta.column_list || [];
+ var vals = meta.rows || [];
+ var mp = {};
+ for (var i = 0; i < cols.length; i++) mp[cols[i]] = vals[i];
+ $('#view_name').text(mp['view_name'] || '');
+ $('#view_definer').text(mp['definer'] || '');
+ $('#view_security_type').text(mp['security_type'] || '');
+ $('#view_check_option').text(mp['check_option'] || '');
+ $('#view_is_updatable').text(mp['is_updatable'] || '');
+ $('#view_character_set_client').text(mp['character_set_client'] || '');
+ // 列信息
+ var desc = result['desc'] || {column_list: [], rows: []};
+ var columns_field = [];
+ $.each(desc.column_list, function (i, column) {
+ columns_field.push({"field": i, "title": column, "sortable": true});
+ });
+ $("#field-list-view").bootstrapTable('destroy').bootstrapTable({
+ escape: true,
+ data: desc.rows,
+ columns: columns_field,
+ striped: true,
+ pagination: true,
+ pageSize: 30,
+ pageList: [30, 50, 100, 500, 1000],
+ locale: 'zh-CN'
+ });
+ renderSqlToPre('#view_definition_sql', result['view_definition']);
+ $('#showModal').modal('show');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
+
+ // 展示触发器详情
+ function showTriggerInfo(ins_name, db_name, trigger_name) {
+ $.ajax({
+ type: "get",
+ url: "/data_dictionary/trigger_info/",
+ dataType: "json",
+ data: {
+ instance_name: ins_name,
+ db_name: db_name,
+ trigger_name: trigger_name,
+ db_type: 'mysql'
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ var result = data.data;
+ showModalBody('trigger');
+ var cols = result.column_list || [];
+ var vals = result.rows || [];
+ var mp = {};
+ for (var i = 0; i < cols.length; i++) mp[cols[i]] = vals[i];
+ $('#trigger_name').text(mp['trigger_name'] || '');
+ $('#trigger_action_timing').text(mp['action_timing'] || '');
+ $('#trigger_event_manipulation').text(mp['event_manipulation'] || '');
+ $('#trigger_event_object_table').text(mp['event_object_table'] || '');
+ $('#trigger_action_orientation').text(mp['action_orientation'] || '');
+ $('#trigger_definer').text(mp['definer'] || '');
+ $('#trigger_created').text(mp['created'] || '');
+ $('#trigger_sql_mode').text(mp['sql_mode'] || '');
+ renderSqlToPre('#trigger_action_statement', mp['action_statement']);
+ $('#showModal').modal('show');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
+
+ // 从 create_sql 结果中提取 SQL 文本
+ function extractCreateSqlText(createSqlRows) {
+ if (!createSqlRows || !createSqlRows.length) return '';
+ var row = createSqlRows[0];
+ if (!row) return '';
+ // SHOW CREATE PROCEDURE/FUNCTION 中第3列为语句
+ // SHOW CREATE EVENT 第4列为语句
+ // 遵循 bootstrapTable field 索引，尝试常见位置
+ if (typeof row === 'object' && !Array.isArray(row)) {
+ // 将对象转为数组
+ var arr = [];
+ for (var k in row) arr.push(row[k]);
+ row = arr;
+ }
+ // 选取最长的字符串字段作为 SQL
+ var best = '';
+ for (var i = 0; i < row.length; i++) {
+ var v = row[i];
+ if (typeof v === 'string' && v.length > best.length) best = v;
+ }
+ return best;
+ }
+
+ // 展示存储过程详情
+ function showProcedureInfo(ins_name, db_name, proc_name) {
+ $.ajax({
+ type: "get",
+ url: "/data_dictionary/procedure_info/",
+ dataType: "json",
+ data: {
+ instance_name: ins_name,
+ db_name: db_name,
+ proc_name: proc_name,
+ db_type: 'mysql'
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ var result = data.data;
+ showModalBody('procedure');
+ var meta = result['meta_data'] || {};
+ var cols = meta.column_list || [];
+ var vals = meta.rows || [];
+ var mp = {};
+ for (var i = 0; i < cols.length; i++) mp[cols[i]] = vals[i];
+ $('#proc_routine_name').text(mp['routine_name'] || '');
+ $('#proc_routine_schema').text(mp['routine_schema'] || '');
+ $('#proc_definer').text(mp['definer'] || '');
+ $('#proc_security_type').text(mp['security_type'] || '');
+ $('#proc_created').text(mp['created'] || '');
+ $('#proc_last_altered').text(mp['last_altered'] || '');
+ $('#proc_sql_mode').text(mp['sql_mode'] || '');
+ $('#proc_routine_comment').text(mp['routine_comment'] || '');
+ renderSqlToPre('#proc_create_sql', extractCreateSqlText(result['create_sql']));
+ $('#showModal').modal('show');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
+
+ // 展示函数详情
+ function showFunctionInfo(ins_name, db_name, func_name) {
+ $.ajax({
+ type: "get",
+ url: "/data_dictionary/function_info/",
+ dataType: "json",
+ data: {
+ instance_name: ins_name,
+ db_name: db_name,
+ func_name: func_name,
+ db_type: 'mysql'
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ var result = data.data;
+ showModalBody('function');
+ var meta = result['meta_data'] || {};
+ var cols = meta.column_list || [];
+ var vals = meta.rows || [];
+ var mp = {};
+ for (var i = 0; i < cols.length; i++) mp[cols[i]] = vals[i];
+ $('#func_routine_name').text(mp['routine_name'] || '');
+ $('#func_return_type').text(mp['return_type'] || '');
+ $('#func_definer').text(mp['definer'] || '');
+ $('#func_routine_schema').text(mp['routine_schema'] || '');
+ $('#func_created').text(mp['created'] || '');
+ $('#func_last_altered').text(mp['last_altered'] || '');
+ $('#func_sql_mode').text(mp['sql_mode'] || '');
+ $('#func_routine_comment').text(mp['routine_comment'] || '');
+ renderSqlToPre('#func_create_sql', extractCreateSqlText(result['create_sql']));
+ $('#showModal').modal('show');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
+
+ // 展示事件详情
+ function showEventInfo(ins_name, db_name, event_name) {
+ $.ajax({
+ type: "get",
+ url: "/data_dictionary/event_info/",
+ dataType: "json",
+ data: {
+ instance_name: ins_name,
+ db_name: db_name,
+ event_name: event_name,
+ db_type: 'mysql'
+ },
+ success: function (data) {
+ if (data.status === 0) {
+ var result = data.data;
+ showModalBody('event');
+ var meta = result['meta_data'] || {};
+ var cols = meta.column_list || [];
+ var vals = meta.rows || [];
+ var mp = {};
+ for (var i = 0; i < cols.length; i++) mp[cols[i]] = vals[i];
+ $('#event_name').text(mp['event_name'] || '');
+ $('#event_status').text(mp['status'] || '');
+ $('#event_event_type').text(mp['event_type'] || '');
+ var interval_desc = '';
+ if (mp['interval_value']) {
+ interval_desc = mp['interval_value'] + ' ' + (mp['interval_field'] || '');
+ } else if (mp['execute_at']) {
+ interval_desc = 'AT ' + mp['execute_at'];
+ }
+ $('#event_interval').text(interval_desc);
+ $('#event_starts').text(mp['starts'] || '');
+ $('#event_ends').text(mp['ends'] || '');
+ $('#event_last_executed').text(mp['last_executed'] || '');
+ $('#event_definer').text(mp['definer'] || '');
+ $('#event_on_completion').text(mp['on_completion'] || '');
+ $('#event_event_comment').text(mp['event_comment'] || '');
+ renderSqlToPre('#event_create_sql', extractCreateSqlText(result['create_sql']));
+ $('#showModal').modal('show');
+ } else {
+ alert(data.msg);
+ }
+ },
+ error: function (XMLHttpRequest, textStatus, errorThrown) {
+ alert(errorThrown);
+ }
+ });
+ }
{% endblock %}
diff --git a/sql/tests.py b/sql/tests.py
index 1bef131182..4e8d47d8b0 100644
--- a/sql/tests.py
+++ b/sql/tests.py
@@ -1964,6 +1964,176 @@ def test_table_info_exception(self, _get_engine):
self.assertEqual(r.status_code, 200)
self.assertDictEqual(json.loads(r.content), {"msg": "test error", "status": 1})
+ # ===== 视图/触发器/存储过程/函数/事件 测试 =====
+ @patch("sql.data_dictionary.get_engine")
+ def test_view_list(self, _get_engine):
+ _get_engine.return_value.get_views_list.return_value = {
+ "v": [["v1", "select 1"]]
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/view_list/", data=data)
+ self.assertEqual(r.status_code, 200)
+ self.assertDictEqual(
+ json.loads(r.content),
+ {"status": 0, "data": {"v": [["v1", "select 1"]]}},
+ )
+
+ def test_view_list_non_mysql(self):
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "oracle",
+ }
+ r = self.client.get(path="/data_dictionary/view_list/", data=data)
+ self.assertDictEqual(
+ json.loads(r.content), {"status": 1, "msg": "仅MySQL支持该功能"}
+ )
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_view_info(self, _get_engine):
+ _get_engine.return_value.get_view_detail.return_value = {
+ "meta_data": {"column_list": [], "rows": []},
+ "desc": {"column_list": [], "rows": []},
+ "view_definition": "select 1",
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "view_name": "v1",
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/view_info/", data=data)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_trigger_list(self, _get_engine):
+ _get_engine.return_value.get_triggers_list.return_value = {
+ "t": [["tg1", "BEFORE INSERT ON t1"]]
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/trigger_list/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_trigger_info(self, _get_engine):
+ _get_engine.return_value.get_trigger_detail.return_value = {
+ "column_list": ["trigger_name"],
+ "rows": ["tg1"],
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "trigger_name": "tg1",
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/trigger_info/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_procedure_list(self, _get_engine):
+ _get_engine.return_value.get_procedures_list.return_value = {
+ "p": [["p1", "cmt"]]
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/procedure_list/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_procedure_info(self, _get_engine):
+ _get_engine.return_value.get_procedure_detail.return_value = {
+ "meta_data": {"column_list": [], "rows": []},
+ "create_sql": [],
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "proc_name": "p1",
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/procedure_info/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_function_list(self, _get_engine):
+ _get_engine.return_value.get_functions_list.return_value = {
+ "f": [["f1", "cmt"]]
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/function_list/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_function_info(self, _get_engine):
+ _get_engine.return_value.get_function_detail.return_value = {
+ "meta_data": {"column_list": [], "rows": []},
+ "create_sql": [],
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "func_name": "f1",
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/function_info/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_event_list(self, _get_engine):
+ _get_engine.return_value.get_events_list.return_value = {
+ "e": [["e1", "ENABLED EVERY 1 DAY"]]
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/event_list/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ @patch("sql.data_dictionary.get_engine")
+ def test_event_info(self, _get_engine):
+ _get_engine.return_value.get_event_detail.return_value = {
+ "meta_data": {"column_list": [], "rows": []},
+ "create_sql": [],
+ }
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "event_name": "e1",
+ "db_type": "mysql",
+ }
+ r = self.client.get(path="/data_dictionary/event_info/", data=data)
+ self.assertEqual(json.loads(r.content)["status"], 0)
+
+ def test_event_info_non_mysql(self):
+ data = {
+ "instance_name": self.ins.instance_name,
+ "db_name": self.db_name,
+ "event_name": "e1",
+ "db_type": "oracle",
+ }
+ r = self.client.get(path="/data_dictionary/event_info/", data=data)
+ self.assertDictEqual(
+ json.loads(r.content), {"status": 1, "msg": "仅MySQL支持该功能"}
+ )
+
def test_export_instance_does_not_exist(self):
"""
测试导出实例不存在
diff --git a/sql/urls.py b/sql/urls.py
index 66142fd6e3..f0773f96d5 100644
--- a/sql/urls.py
+++ b/sql/urls.py
@@ -122,6 +122,16 @@
path("data_dictionary/", views.data_dictionary),
path("data_dictionary/table_list/", data_dictionary.table_list),
path("data_dictionary/table_info/", data_dictionary.table_info),
+ path("data_dictionary/view_list/", data_dictionary.view_list),
+ path("data_dictionary/view_info/", data_dictionary.view_info),
+ path("data_dictionary/trigger_list/", data_dictionary.trigger_list),
+ path("data_dictionary/trigger_info/", data_dictionary.trigger_info),
+ path("data_dictionary/procedure_list/", data_dictionary.procedure_list),
+ path("data_dictionary/procedure_info/", data_dictionary.procedure_info),
+ path("data_dictionary/function_list/", data_dictionary.function_list),
+ path("data_dictionary/function_info/", data_dictionary.function_info),
+ path("data_dictionary/event_list/", data_dictionary.event_list),
+ path("data_dictionary/event_info/", data_dictionary.event_info),
path("data_dictionary/export/", data_dictionary.export),
path("param/list/", instance.param_list),
path("param/history/", instance.param_history),