解决DML语句包含头部注释会绕过自动审批判断的问题 fix #949

移除sqlparse解析和Inception二次检测，提高判断效率

sql/utils/sql_review.py

@@ -1,12 +1,13 @@
 import datetime
+import json
 import re
-import sqlparse
 from django.db import transaction
 
+from sql.engines.models import ReviewResult
 from sql.models import SqlWorkflow
 from common.config import SysConfig
 from sql.utils.resource_group import user_groups
-from sql.engines import get_engine
+from sql.utils.sql_utils import remove_comments
 
 
 def is_auto_review(workflow_id):
@@ -22,31 +23,26 @@ def is_auto_review(workflow_id):
     if workflow.instance.db_type == 'mysql' and workflow.instance.instance_tag.filter(
             tag_code__in=auto_review_tags).exists():
         # 获取正则表达式
-        auto_review_regex = SysConfig().get('auto_review_regex',
-                                            '^alter|^create|^drop|^truncate|^rename|^delete')
+        auto_review_regex = SysConfig().get(
+            'auto_review_regex', '^alter|^create|^drop|^truncate|^rename|^delete')
         p = re.compile(auto_review_regex, re.I)
 
         # 判断是否匹配到需要手动审核的语句
         auto_review = True
-        sql_content = workflow.sqlworkflowcontent.sql_content
-        for statement in sqlparse.split(sql_content):
-            # 删除注释语句
-            statement = sqlparse.format(statement, strip_comments=True)
-            if p.match(statement.strip()):
+        all_affected_rows = 0
+        review_content = workflow.sqlworkflowcontent.review_content
+        for review_row in json.loads(review_content):
+            review_result = ReviewResult(**review_row)
+            # 去除SQL注释 https://github.com/hhyo/Archery/issues/949
+            sql = remove_comments(review_result.sql)
+            # 正则匹配
+            if p.match(sql):
                 auto_review = False
                 break
-            if auto_review:
-                # 更新影响行数加测,总语句影响行数超过指定数量则需要人工审核
-                review_engine = get_engine(instance=workflow.instance)
-                inception_review = review_engine.execute_check(db_name=workflow.db_name, sql=sql_content).to_dict()
-                all_affected_rows = 0
-                for review_result in inception_review:
-                    sql = review_result.get('sql', '')
-                    affected_rows = review_result.get('affected_rows', 0)
-                    if re.match(r"^update", sql.strip().lower()):
-                        all_affected_rows = all_affected_rows + int(affected_rows)
-                if int(all_affected_rows) > int(SysConfig().get('auto_review_max_update_rows', 50)):
-                    auto_review = False
+            # 影响行数加测, 总语句影响行数超过指定数量则需要人工审核
+            all_affected_rows += int(review_result.affected_rows)
+        if all_affected_rows > int(SysConfig().get('auto_review_max_update_rows', 50)):
+            auto_review = False
     else:
         auto_review = False
     return auto_review

sql/utils/tests.py

@@ -209,9 +209,8 @@ def test_auto_review_gt_max_update_rows(self, _get_engine, _execute_check):
         r = is_auto_review(self.wfc1.workflow_id)
         self.assertFalse(r)
 
-    @patch('sql.engines.mysql.MysqlEngine.execute_check')
     @patch('sql.engines.get_engine')
-    def test_auto_review_true(self, _get_engine, _execute_check):
+    def test_auto_review_true(self, _get_engine):
         """
         测试自动审批通过的判定条件，
         :return:
@@ -222,21 +221,20 @@ def test_auto_review_true(self, _get_engine, _execute_check):
         self.sys_config.set('auto_review_max_update_rows', '2')  # update影响行数大于2需要审批
         self.sys_config.set('auto_review_tag', 'GA')  # 仅GA开启自动审批
         self.sys_config.get_all_config()
-        # 修改工单为update
+        # 修改工单为update，mock返回值，update影响行数=3
         self.wfc1.sql_content = "update table users set email='';"
-        self.wfc1.save(update_fields=('sql_content',))
-        # 修改工单实例标签
-        tag, is_created = InstanceTag.objects.get_or_create(tag_code='GA', defaults={'tag_name': '生产环境', 'active': True})
-        self.wf1.instance.instance_tag.add(tag)
-        # mock返回值，update影响行数=3
-        _execute_check.return_value.to_dict.return_value = [
+        self.wfc1.review_content = json.dumps([
             {"id": 1, "stage": "CHECKED", "errlevel": 0, "stagestatus": "Audit completed", "errormessage": "None",
              "sql": "use archer_test", "affected_rows": 0, "sequence": "'0_0_0'", "backup_dbname": "None",
              "execute_time": "0", "sqlsha1": "", "actual_affected_rows": 'null'},
             {"id": 2, "stage": "CHECKED", "errlevel": 0, "stagestatus": "Audit completed", "errormessage": "None",
              "sql": "update table users set email=''", "affected_rows": 1, "sequence": "'0_0_1'",
              "backup_dbname": "mysql_3306_archer_test", "execute_time": "0", "sqlsha1": "",
-             "actual_affected_rows": 'null'}]
+             "actual_affected_rows": 'null'}])
+        self.wfc1.save(update_fields=('sql_content','review_content'))
+        # 修改工单实例标签
+        tag, is_created = InstanceTag.objects.get_or_create(tag_code='GA', defaults={'tag_name': '生产环境', 'active': True})
+        self.wf1.instance.instance_tag.add(tag)
         r = is_auto_review(self.wfc1.workflow_id)
         self.assertTrue(r)