增加过滤数据库选项

sql/engines/__init__.py

@@ -28,7 +28,8 @@ def __init__(self, instance=None):
             self.mode = instance.mode
             self.is_ssl = instance.is_ssl
             self.is_ignore_certificate_error = instance.is_ignore_certificate_error
-            self.allow_db_name_list = instance.allow_db_name_list
+            self.show_db_name_regex = instance.show_db_name_regex
+            self.denied_db_name_regex = instance.denied_db_name_regex
 
             # 判断如果配置了隧道则连接隧道，只测试了MySQL
             if self.instance.tunnel:

sql/engines/mongo.py

@@ -14,7 +14,7 @@
 from bson.objectid import ObjectId
 from bson.int64 import Int64
 
-from sql.utils.sql_utils import filter_db_list
+from sql.utils.sql_utils import filter_denied_db_list, filter_show_db_list
 
 from . import EngineBase
 from .models import ResultSet, ReviewSet, ReviewResult
@@ -843,7 +843,9 @@ def get_all_databases(self):
             db_list = conn.list_database_names()
         except OperationFailure:
             db_list = [self.db_name]
-        result.rows = filter_db_list(db_list, self.allow_db_name_list)
+        db_list = filter_show_db_list(db_list, self.show_db_name_regex)
+        db_list = filter_denied_db_list(db_list, self.denied_db_name_regex)
+        result.rows = db_list
         return result
 
     def get_all_tables(self, db_name, **kwargs):

sql/engines/mysql.py

@@ -12,7 +12,7 @@
 from schemaobject.connection import build_database_url
 
 from sql.engines.goinception import GoInceptionEngine
-from sql.utils.sql_utils import get_syntax_type, remove_comments
+from sql.utils.sql_utils import filter_denied_db_list, filter_show_db_list, get_syntax_type, remove_comments
 from . import EngineBase
 from .models import ResultSet, ReviewResult, ReviewSet
 from sql.utils.data_masking import data_masking
@@ -194,6 +194,8 @@ def get_all_databases(self):
         db_list = [
             row[0] for row in result.rows if row[0] not in self.forbidden_databases
         ]
+        db_list = filter_show_db_list(db_list, self.show_db_name_regex)
+        db_list = filter_denied_db_list(db_list, self.denied_db_name_regex)
         result.rows = db_list
         return result
 

sql/engines/redis.py

@@ -15,7 +15,7 @@
 import traceback
 
 from common.utils.timer import FuncTimer
-from sql.utils.sql_utils import filter_db_list
+from sql.utils.sql_utils import filter_denied_db_list, filter_show_db_list
 from . import EngineBase
 from .models import ResultSet, ReviewSet, ReviewResult
 
@@ -86,7 +86,8 @@ def get_all_databases(self, **kwargs):
             rows = max(dbs + [15]) + 1
 
         db_list = [str(x) for x in range(int(rows))]
-        db_list = filter_db_list(db_list, self.allow_db_name_list)
+        db_list = filter_show_db_list(db_list, self.show_db_name_regex)
+        db_list = filter_denied_db_list(db_list, self.denied_db_name_regex)
         result.rows = db_list
         return result
 

sql/engines/tests.py

@@ -328,10 +328,10 @@ def test_get_all_databases(self, _config_get):
         self.assertListEqual(dbs.rows, ["0", "1", "2", "3"])
 
     @patch("redis.Redis.config_get", return_value={"databases": 20})
-    def test_get_all_databases_with_allow_db_name_list(self, _config_get):
+    def test_get_all_databases_with_show_db_name_regex(self, _config_get):
         """获取数据库列表。"""
         new_engine = RedisEngine(instance=self.ins)
-        new_engine.allow_db_name_list = "0,4,6~9,"
+        new_engine.show_db_name_regex = "0,4,6~9,"
         dbs = new_engine.get_all_databases()
         # 预期结果：过滤后的数据库列表应只包括允许的数据库名
         expected_result = ["0", "4", "6", "7", "8", "9"]

sql/models.py

@@ -207,13 +207,21 @@ class Instance(models.Model):
     is_ssl = models.BooleanField("是否启用SSL", default=False)
     is_ignore_certificate_error = models.BooleanField("是否忽略证书错误", default=False)
     db_name = models.CharField("数据库", max_length=64, default="", blank=True)
-    allow_db_name_list = models.CharField(
-        "允许显示的数据库列表",
+    show_db_name_regex = models.CharField(
+        "显示的数据库列表正则",
         max_length=1024,
         default="",
         blank=True,
-        help_text="多个以逗号隔开，支持*和~。MySQL示例：test_db,dmp_db,za_ag。Redis示例: 0,4,20~26",
+        help_text="正则表达式。示例：^(test_db|dmp_db|za.*)$。Redis示例: ^(0|4|6|11|12|13)$",
     )
+    denied_db_name_regex = models.CharField(
+        "隐藏的数据库列表正则",
+        max_length=1024,
+        default="",
+        blank=True,
+        help_text="正则表达式。隐藏大于显示，此规则优先。",
+    )
+
     charset = models.CharField("字符集", max_length=20, default="", blank=True)
     service_name = models.CharField(
         "Oracle service name", max_length=50, null=True, blank=True

sql/utils/sql_utils.py

@@ -382,58 +382,78 @@ def get_exec_sqlitem_list(reviewResult, db_name):
     return list
 
 
-def filter_db_list(db_list, allow_db_name_list, key="value"):
+def filter_show_db_list(db_list, show_db_name_regex, key="value"):
     """
     根据配置的数据库列表过滤数据库名称列表。
 
     :param db_list: 待过滤的数据库名称列表，可能是字符串列表或字典列表。
     示例数据：
     1. db_list=[{"value": 0, "text": 0, "value": 1, "text": 1}]
     2. db_list=["a_db","b_db"]
-    :param allow_db_name_list: 配置的数据库显示列表。支持通配符 * 和范围 ~。
+    :param show_db_name_regex: 配置的数据库显示正则。
     :param key: 当 db_list 包含字典时，指定用于匹配的键。默认值为 'value'。
     :return: 过滤后的数据库名称列表或字典列表。
     """
-    if not allow_db_name_list:
-        return db_list  # 如果没有指定 allow_db_name_list，则返回原始 db_list
+    if not show_db_name_regex:
+        return db_list  # 如果没有指定 show_db_name_regex，返回原始 db_list
 
-    # 将通配符模式字符串转换为列表
-    allowed_patterns = allow_db_name_list.split(",")
+    try:
+        allowed_regex = re.compile(show_db_name_regex)  # 编译正则表达式
+    except re.error:
+        raise ValueError(f"正则表达式解析异常: {show_db_name_regex}")
 
-    allowed_regexes = []
-    for pattern in allowed_patterns:
-        pattern = pattern.strip()
-        if not pattern:
-            continue
-        if "~" in pattern:
-            # 如果模式包含范围 "~"，解析范围并生成数字列表
-            try:
-                start, end = map(int, pattern.split("~"))
-                # 将生成的数字添加到允许列表的正则表达式列表
-                allowed_regexes.extend(
-                    [re.compile(f"^{db}$") for db in range(start, end + 1)]
-                )
-            except ValueError:
-                # 如果范围内的值不是有效的整数
-                allowed_regexes.append(re.compile(f"^{pattern}$".replace("*", ".*")))
-        else:
-            # 转义特殊字符,并将通配符模式转换为正则表达式
-            escaped_pattern = re.escape(pattern).replace(r"\*", ".*")
-            allowed_regexes.append(re.compile(f"^{escaped_pattern}$"))
+    filtered_list = []
+
+    # 根据类型处理 db_list
+    if all(isinstance(db, dict) for db in db_list):
+        # 如果 db_list 是一个字典的列表
+        for db in db_list:
+            db_value = str(db.get(key, ""))  # 确保 db_value 是字符串类型
+            if allowed_regex.match(db_value):
+                filtered_list.append(db)
+    else:
+        # 如果 db_list 是一个字符串的列表
+        for db in db_list:
+            if allowed_regex.match(db):
+                filtered_list.append(db)
+
+    return filtered_list
+
+
+def filter_denied_db_list(db_list, denied_db_name_regex, key="value"):
+    """
+    根据配置的数据库列表过滤数据库名称列表。
+
+    :param db_list: 待过滤的数据库名称列表，可能是字符串列表或字典列表。
+    示例数据：
+    1. db_list=[{"value": 0, "text": 0, "value": 1, "text": 1}]
+    2. db_list=["a_db","b_db"]
+    :param denied_db_name_regex: 配置的数据库隐藏正则。
+    :param key: 当 db_list 包含字典时，指定用于匹配的键。默认值为 'value'。
+    :return: 过滤后的数据库名称列表或字典列表。
+    """
+    if not denied_db_name_regex:
+        return db_list
+    try:
+        denied_regex = re.compile(denied_db_name_regex)
+    except re.error:
+        raise ValueError(f"正则表达式解析异常: {denied_db_name_regex}")
 
     filtered_list = []
 
     # 根据类型处理 db_list
     if all(isinstance(db, dict) for db in db_list):
         # 如果 db_list 是一个字典的列表
         for db in db_list:
-            db_value = db.get(key, "")
-            if any(regex.match(db_value) for regex in allowed_regexes):
+            db_value = str(db.get(key, ""))  # 确保 db_value 是字符串类型
+            if not denied_regex.match(
+                db_value
+            ):  # 只有不匹配隐藏正则表达式的才添加到列表中
                 filtered_list.append(db)
     else:
         # 如果 db_list 是一个字符串的列表
         for db in db_list:
-            if any(regex.match(db) for regex in allowed_regexes):
+            if not denied_regex.match(db):  # 只有不匹配隐藏正则表达式的才添加到列表中
                 filtered_list.append(db)
 
     return filtered_list

src/init_sql/v1.12.0.sql

@@ -1,3 +1,5 @@
 
 ALTER TABLE sql_instance ADD is_ignore_certificate_error tinyint(1) DEFAULT 0  COMMENT '是否忽略证书错误。1：忽略。0：不忽略，需要验证';
-ALTER TABLE sql_instance ADD allow_db_name_list varchar(1024) DEFAULT ''  COMMENT '允许显示的数据库列表，多个以逗号隔开，支持*和~。';
+ALTER TABLE sql_instance ADD show_db_name_regex varchar(1024) DEFAULT ''  COMMENT '显示的数据库列表正则';
+ALTER TABLE sql_instance ADD denied_db_name_regex varchar(1024) DEFAULT ''  COMMENT '隐藏的数据库列表正则';
+