diff --git a/common/templates/config.html b/common/templates/config.html
index c66075a0f6..511dd607dd 100755
--- a/common/templates/config.html
+++ b/common/templates/config.html
@@ -27,12 +27,15 @@ <h5 class="control-label text-bold">工单类型：</h5>
                             <select id="workflow_type" name="group"
                                     class="selectpicker show-tick form-control bs-select-hidden"
                                     data-name="工单类型" data-placeholder="请选择工单类型:" required />
-                                <option value="{{ WorkflowDict.workflow_type.sqlreview }}" selected="selected">
-                                    {{ WorkflowDict.workflow_type.sqlreview_display }}</option>
-                                <option value="{{ WorkflowDict.workflow_type.query }}">
-                                    {{ WorkflowDict.workflow_type.query_display }}</option>
-                                <option value="{{ WorkflowDict.workflow_type.archive }}">
-                                    {{ WorkflowDict.workflow_type.archive_display }}</option>
+                                {% for c in workflow_choices %}
+                                {% if forloop.first %}
+                                <option value="{{ c.value }}" selected="selected">
+                                {{ c.label }}</option>
+                                {% else %}
+                                <option value="{{ c.value }}">
+                                {{ c.label }}</option>
+                                {% endif %}
+                                {% endfor %}
                             </select>
                         </div>
                         <h5 class="control-label text-bold">组：</h5>
diff --git a/common/utils/const.py b/common/utils/const.py
index 96d4c363a5..2a1f336cbd 100644
--- a/common/utils/const.py
+++ b/common/utils/const.py
@@ -1,4 +1,4 @@
-# -*- coding: UTF-8 -*-
+from django.db import models
 
 
 class Const(object):
@@ -10,28 +10,17 @@ class Const(object):
     }
 
 
-class WorkflowDict:
-    # 工作流申请类型，1.query,2.SQL上线申请
-    workflow_type = {
-        "query": 1,
-        "query_display": "查询权限申请",
-        "sqlreview": 2,
-        "sqlreview_display": "SQL上线申请",
-        "archive": 3,
-        "archive_display": "数据归档申请",
-    }
+class WorkflowType(models.IntegerChoices):
+    QUERY = 1, "查询权限申请"
+    SQL_REVIEW = 2, "SQL上线申请"
+    ARCHIVE = 3, "数据归档申请"
 
-    # 工作流状态，0.待审核 1.审核通过 2.审核不通过 3.审核取消
-    workflow_status = {
-        "audit_wait": 0,
-        "audit_wait_display": "待审核",
-        "audit_success": 1,
-        "audit_success_display": "审核通过",
-        "audit_reject": 2,
-        "audit_reject_display": "审核不通过",
-        "audit_abort": 3,
-        "audit_abort_display": "审核取消",
-    }
+
+class WorkflowStatus(models.IntegerChoices):
+    WAITING = 0, "待审核"
+    PASSED = 1, "审核通过"
+    REJECTED = 2, "审核不通过"
+    ABORTED = 3, "审核取消"
 
 
 class SQLTuning:
diff --git a/common/workflow.py b/common/workflow.py
index 2ca3edcfbc..c7029ef1e6 100644
--- a/common/workflow.py
+++ b/common/workflow.py
@@ -2,7 +2,7 @@
 from django.contrib.auth.models import Group
 from django.http import HttpResponse
 
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus
 from common.utils.extend_json_encoder import ExtendJSONEncoder, ExtendJSONEncoderFTime
 from sql.models import WorkflowAudit, WorkflowLog
 from sql.utils.resource_group import user_groups
@@ -31,7 +31,7 @@ def lists(request):
     # 只返回所在资源组当前待自己审核的数据
     workflow_audit = WorkflowAudit.objects.filter(
         workflow_title__icontains=search,
-        current_status=WorkflowDict.workflow_status["audit_wait"],
+        current_status=WorkflowStatus.WAITING,
         group_id__in=group_ids,
         current_audit__in=auth_group_ids,
     )
diff --git a/sql/archiver.py b/sql/archiver.py
index c054921d42..61bea7dd25 100644
--- a/sql/archiver.py
+++ b/sql/archiver.py
@@ -22,7 +22,7 @@
 from django.urls import reverse
 from django_q.tasks import async_task
 
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from common.utils.extend_json_encoder import ExtendJSONEncoder
 from common.utils.timer import FuncTimer
 from sql.engines import get_engine
@@ -171,9 +171,7 @@ def archive_apply(request):
 
     # 获取资源组和审批信息
     res_group = ResourceGroup.objects.get(group_name=group_name)
-    audit_auth_groups = Audit.settings(
-        res_group.group_id, WorkflowDict.workflow_type["archive"]
-    )
+    audit_auth_groups = Audit.settings(res_group.group_id, WorkflowType.ARCHIVE)
     if not audit_auth_groups:
         return JsonResponse({"status": 1, "msg": "审批流程不能为空，请先配置审批流程", "data": {}})
 
@@ -195,16 +193,14 @@ def archive_apply(request):
                 mode=mode,
                 no_delete=no_delete,
                 sleep=sleep,
-                status=WorkflowDict.workflow_status["audit_wait"],
+                status=WorkflowStatus.WAITING,
                 state=False,
                 user_name=user.username,
                 user_display=user.display,
             )
             archive_id = archive_info.id
             # 调用工作流插入审核信息
-            audit_result, audit_detail = Audit.add(
-                WorkflowDict.workflow_type["archive"], archive_id
-            )
+            audit_result, audit_detail = Audit.add(WorkflowType.ARCHIVE, archive_id)
     except Exception as msg:
         logger.error(traceback.format_exc())
         result["status"] = 1
@@ -213,7 +209,7 @@ def archive_apply(request):
         result = audit_result
         # 消息通知
         workflow_audit = Audit.detail_by_workflow_id(
-            workflow_id=archive_id, workflow_type=WorkflowDict.workflow_type["archive"]
+            workflow_id=archive_id, workflow_type=WorkflowType.ARCHIVE
         )
         async_task(
             notify_for_audit,
@@ -250,7 +246,7 @@ def archive_audit(request):
         with transaction.atomic():
             workflow_audit = Audit.detail_by_workflow_id(
                 workflow_id=archive_id,
-                workflow_type=WorkflowDict.workflow_type["archive"],
+                workflow_type=WorkflowType.ARCHIVE,
             )
             audit_id = workflow_audit.audit_id
 
@@ -262,9 +258,7 @@ def archive_audit(request):
             ArchiveConfig(
                 id=archive_id,
                 status=audit_status,
-                state=True
-                if audit_status == WorkflowDict.workflow_status["audit_success"]
-                else False,
+                state=True if audit_status == WorkflowStatus.PASSED else False,
             ).save(update_fields=["status", "state"])
     except Exception as msg:
         logger.error(traceback.format_exc())
@@ -298,11 +292,11 @@ def add_archive_task(archive_ids=None):
         archive_cnf_list = ArchiveConfig.objects.filter(
             id__in=archive_ids,
             state=True,
-            status=WorkflowDict.workflow_status["audit_success"],
+            status=WorkflowStatus.PASSED,
         )
     else:
         archive_cnf_list = ArchiveConfig.objects.filter(
-            state=True, status=WorkflowDict.workflow_status["audit_success"]
+            state=True, status=WorkflowStatus.PASSED
         )
 
     # 添加task任务
diff --git a/sql/models.py b/sql/models.py
index 0547d0cb64..20983a3fa8 100755
--- a/sql/models.py
+++ b/sql/models.py
@@ -5,6 +5,8 @@
 from django.utils.translation import gettext as _
 from mirage.crypto import Crypto
 
+from common.utils.const import WorkflowStatus, WorkflowType
+
 
 class ResourceGroup(models.Model):
     """
@@ -304,10 +306,6 @@ class Meta:
         verbose_name_plural = "SQL工单内容"
 
 
-workflow_type_choices = ((1, _("sql_query")), (2, _("sql_review")))
-workflow_status_choices = ((0, "待审核"), (1, "审核通过"), (2, "审核不通过"), (3, "审核取消"))
-
-
 class WorkflowAudit(models.Model):
     """
     工作流审核状态表
@@ -317,13 +315,13 @@ class WorkflowAudit(models.Model):
     group_id = models.IntegerField("组ID")
     group_name = models.CharField("组名称", max_length=100)
     workflow_id = models.BigIntegerField("关联业务id")
-    workflow_type = models.IntegerField("申请类型", choices=workflow_type_choices)
+    workflow_type = models.IntegerField("申请类型", choices=WorkflowType.choices)
     workflow_title = models.CharField("申请标题", max_length=50)
     workflow_remark = models.CharField("申请备注", default="", max_length=140, blank=True)
     audit_auth_groups = models.CharField("审批权限组列表", max_length=255)
     current_audit = models.CharField("当前审批权限组", max_length=20)
     next_audit = models.CharField("下级审批权限组", max_length=20)
-    current_status = models.IntegerField("审核状态", choices=workflow_status_choices)
+    current_status = models.IntegerField("审核状态", choices=WorkflowStatus.choices)
     create_user = models.CharField("申请人", max_length=30)
     create_user_display = models.CharField("申请人中文名", max_length=50, default="")
     create_time = models.DateTimeField("申请时间", auto_now_add=True)
@@ -349,7 +347,7 @@ class WorkflowAuditDetail(models.Model):
     audit_id = models.IntegerField("审核主表id")
     audit_user = models.CharField("审核人", max_length=30)
     audit_time = models.DateTimeField("审核时间")
-    audit_status = models.IntegerField("审核状态", choices=workflow_status_choices)
+    audit_status = models.IntegerField("审核状态", choices=WorkflowStatus.choices)
     remark = models.CharField("审核备注", default="", max_length=1000)
     sys_time = models.DateTimeField("系统时间", auto_now=True)
 
@@ -371,7 +369,7 @@ class WorkflowAuditSetting(models.Model):
     audit_setting_id = models.AutoField(primary_key=True)
     group_id = models.IntegerField("组ID")
     group_name = models.CharField("组名称", max_length=100)
-    workflow_type = models.IntegerField("审批类型", choices=workflow_type_choices)
+    workflow_type = models.IntegerField("审批类型", choices=WorkflowType.choices)
     audit_auth_groups = models.CharField("审批权限组列表", max_length=255)
     create_time = models.DateTimeField(auto_now_add=True)
     sys_time = models.DateTimeField(auto_now=True)
@@ -446,7 +444,7 @@ class QueryPrivilegesApply(models.Model):
         ),
         default=0,
     )
-    status = models.IntegerField("审核状态", choices=workflow_status_choices)
+    status = models.IntegerField("审核状态", choices=WorkflowStatus.choices)
     audit_auth_groups = models.CharField("审批权限组列表", max_length=255)
     create_time = models.DateTimeField(auto_now_add=True)
     sys_time = models.DateTimeField(auto_now=True)
@@ -720,7 +718,7 @@ class ArchiveConfig(models.Model):
     no_delete = models.BooleanField("是否保留源数据")
     sleep = models.IntegerField("归档limit行后的休眠秒数", default=1)
     status = models.IntegerField(
-        "审核状态", choices=workflow_status_choices, blank=True, default=1
+        "审核状态", choices=WorkflowStatus.choices, blank=True, default=1
     )
     state = models.BooleanField("是否启用归档", default=True)
     user_name = models.CharField("申请人", max_length=30, blank=True, default="")
diff --git a/sql/notify.py b/sql/notify.py
index f812084b4d..b478bba9bc 100755
--- a/sql/notify.py
+++ b/sql/notify.py
@@ -13,7 +13,7 @@
 from django.contrib.auth.models import Group
 
 from common.config import SysConfig
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from common.utils.sendmsg import MsgSender
 from sql.models import (
     QueryPrivilegesApply,
@@ -162,8 +162,8 @@ def render_audit(self):
         )
         # workflow content, 即申请通过后要执行什么东西
         # 执行的 SQL 语句, 授权的范围
-        if workflow_type == WorkflowDict.workflow_type["query"]:
-            workflow_type_display = WorkflowDict.workflow_type["query_display"]
+        if workflow_type == WorkflowType.QUERY:
+            workflow_type_display = WorkflowType.QUERY.label
             workflow_detail = QueryPrivilegesApply.objects.get(apply_id=workflow_id)
             instance = workflow_detail.instance.instance_name
             db_name = workflow_detail.db_list
@@ -178,8 +178,8 @@ def render_audit(self):
             workflow_content += (
                 f"""授权截止时间：{auth_ends_at}\n结果集：{workflow_detail.limit_num}\n"""
             )
-        elif workflow_type == WorkflowDict.workflow_type["sqlreview"]:
-            workflow_type_display = WorkflowDict.workflow_type["sqlreview_display"]
+        elif workflow_type == WorkflowType.SQL_REVIEW:
+            workflow_type_display = WorkflowType.SQL_REVIEW.label
             workflow_detail = SqlWorkflow.objects.get(pk=workflow_id)
             instance = workflow_detail.instance.instance_name
             db_name = workflow_detail.db_name
@@ -188,8 +188,8 @@ def render_audit(self):
                 "\n",
                 workflow_detail.sqlworkflowcontent.sql_content[0:500].replace("\r", ""),
             )
-        elif workflow_type == WorkflowDict.workflow_type["archive"]:
-            workflow_type_display = WorkflowDict.workflow_type["archive_display"]
+        elif workflow_type == WorkflowType.ARCHIVE:
+            workflow_type_display = WorkflowType.ARCHIVE.label
             workflow_detail = ArchiveConfig.objects.get(pk=workflow_id)
             instance = workflow_detail.src_instance.instance_name
             db_name = workflow_detail.src_db_name
@@ -201,7 +201,7 @@ def render_audit(self):
         else:
             raise Exception("工单类型不正确")
         # 渲染提醒内容, 包括工单的所有信息, 申请人, 审批流等
-        if status == WorkflowDict.workflow_status["audit_wait"]:  # 申请阶段
+        if status == WorkflowStatus.WAITING:  # 申请阶段
             msg_title = "[{}]新的工单申请#{}".format(workflow_type_display, audit_id)
             # 接收人，发送给该资源组内对应权限组所有的用户
             auth_group_names = Group.objects.get(id=self.audit.current_audit).name
@@ -228,7 +228,7 @@ def render_audit(self):
                 workflow_url,
                 workflow_content,
             )
-        elif status == WorkflowDict.workflow_status["audit_success"]:  # 审核通过
+        elif status == WorkflowStatus.PASSED:  # 审核通过
             msg_title = "[{}]工单审核通过#{}".format(workflow_type_display, audit_id)
             # 接收人，仅发送给申请人
             msg_to = [Users.objects.get(username=self.audit.create_user)]
@@ -244,7 +244,7 @@ def render_audit(self):
                 workflow_url,
                 workflow_content,
             )
-        elif status == WorkflowDict.workflow_status["audit_reject"]:  # 审核驳回
+        elif status == WorkflowStatus.REJECTED:  # 审核驳回
             msg_title = "[{}]工单被驳回#{}".format(workflow_type_display, audit_id)
             # 接收人，仅发送给申请人
             msg_to = [Users.objects.get(username=self.audit.create_user)]
@@ -257,7 +257,7 @@ def render_audit(self):
                 workflow_url,
                 re.sub("[\r\n\f]{2,}", "\n", self.audit_detail.remark),
             )
-        elif status == WorkflowDict.workflow_status["audit_abort"]:  # 审核取消，通知所有审核人
+        elif status == WorkflowStatus.ABORTED:  # 审核取消，通知所有审核人
             msg_title = "[{}]提交人主动终止工单#{}".format(workflow_type_display, audit_id)
             # 接收人，发送给该资源组内对应权限组所有的用户
             auth_group_names = [
@@ -293,7 +293,7 @@ def render_execute(self):
             base_url=base_url, audit_id=audit_id
         )
         msg_title = (
-            f"[{WorkflowDict.workflow_type['sqlreview_display']}]工单"
+            f"[{WorkflowType.SQL_REVIEW.label}]工单"
             f"{self.workflow.get_status_display()}#{audit_id}"
         )
         preview = re.sub(
diff --git a/sql/query_privileges.py b/sql/query_privileges.py
index 90c6d73f84..f00aa5d55b 100644
--- a/sql/query_privileges.py
+++ b/sql/query_privileges.py
@@ -20,7 +20,7 @@
 from django_q.tasks import async_task
 
 from common.config import SysConfig
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from common.utils.extend_json_encoder import ExtendJSONEncoder
 from sql.engines.goinception import GoInceptionEngine
 from sql.models import QueryPrivilegesApply, QueryPrivileges, Instance, ResourceGroup
@@ -267,15 +267,13 @@ def query_priv_apply(request):
                 title=title,
                 group_id=group_id,
                 group_name=group_name,
-                audit_auth_groups=Audit.settings(
-                    group_id, WorkflowDict.workflow_type["query"]
-                ),
+                audit_auth_groups=Audit.settings(group_id, WorkflowType.QUERY),
                 user_name=user.username,
                 user_display=user.display,
                 instance=ins,
                 priv_type=int(priv_type),
                 valid_date=valid_date,
-                status=WorkflowDict.workflow_status["audit_wait"],
+                status=WorkflowStatus.WAITING,
                 limit_num=limit_num,
             )
             if int(priv_type) == 1:
@@ -288,7 +286,7 @@ def query_priv_apply(request):
             apply_id = applyinfo.apply_id
 
             # 调用工作流插入审核信息,查询权限申请workflow_type=1
-            audit_result = Audit.add(WorkflowDict.workflow_type["query"], apply_id)
+            audit_result = Audit.add(WorkflowType.QUERY, apply_id)
             if audit_result["status"] == 0:
                 # 更新业务表审核状态,判断是否插入权限信息
                 _query_apply_audit_call_back(
@@ -302,7 +300,7 @@ def query_priv_apply(request):
         result = audit_result
         # 消息通知
         workflow_audit = Audit.detail_by_workflow_id(
-            workflow_id=apply_id, workflow_type=WorkflowDict.workflow_type["query"]
+            workflow_id=apply_id, workflow_type=WorkflowType.QUERY
         )
         async_task(
             notify_for_audit,
@@ -440,7 +438,7 @@ def query_priv_audit(request):
     try:
         with transaction.atomic():
             audit_id = Audit.detail_by_workflow_id(
-                workflow_id=apply_id, workflow_type=WorkflowDict.workflow_type["query"]
+                workflow_id=apply_id, workflow_type=WorkflowType.QUERY
             ).audit_id
 
             # 调用工作流接口审核
@@ -450,7 +448,7 @@ def query_priv_audit(request):
 
             # 按照审核结果更新业务表审核状态
             audit_detail = Audit.detail(audit_id)
-            if audit_detail.workflow_type == WorkflowDict.workflow_type["query"]:
+            if audit_detail.workflow_type == WorkflowType.QUERY:
                 # 更新业务表审核状态,插入权限信息
                 _query_apply_audit_call_back(
                     audit_detail.workflow_id, audit_result["data"]["workflow_status"]
@@ -578,7 +576,7 @@ def _query_apply_audit_call_back(apply_id, workflow_status):
     apply_info.status = workflow_status
     apply_info.save()
     # 审核通过插入权限信息，批量插入，减少性能消耗
-    if workflow_status == WorkflowDict.workflow_status["audit_success"]:
+    if workflow_status == WorkflowStatus.PASSED:
         apply_queryset = QueryPrivilegesApply.objects.get(apply_id=apply_id)
         # 库权限
 
diff --git a/sql/sql_workflow.py b/sql/sql_workflow.py
index 0ac8f7c899..a05e8052e9 100644
--- a/sql/sql_workflow.py
+++ b/sql/sql_workflow.py
@@ -14,7 +14,7 @@
 from django_q.tasks import async_task
 
 from common.config import SysConfig
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from common.utils.extend_json_encoder import ExtendJSONEncoder
 from sql.engines import get_engine
 from sql.engines.models import ReviewResult, ReviewSet
@@ -250,21 +250,18 @@ def passed(request):
             # 调用工作流接口审核
             workflow_audit = Audit.detail_by_workflow_id(
                 workflow_id=workflow_id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             )
             audit_id = workflow_audit.audit_id
             audit_result, audit_detail = Audit.audit(
                 audit_id,
-                WorkflowDict.workflow_status["audit_success"],
+                WorkflowStatus.PASSED,
                 user.username,
                 audit_remark,
             )
 
             # 按照审核结果更新业务表审核状态
-            if (
-                audit_result["data"]["workflow_status"]
-                == WorkflowDict.workflow_status["audit_success"]
-            ):
+            if audit_result["data"]["workflow_status"] == WorkflowStatus.PASSED:
                 # 将流程状态修改为审核通过
                 SqlWorkflow(id=workflow_id, status="workflow_review_pass").save(
                     update_fields=["status"]
@@ -319,7 +316,7 @@ def execute(request):
         return render(request, "error.html", context)
     # 获取审核信息
     audit_id = Audit.detail_by_workflow_id(
-        workflow_id=workflow_id, workflow_type=WorkflowDict.workflow_type["sqlreview"]
+        workflow_id=workflow_id, workflow_type=WorkflowType.SQL_REVIEW
     ).audit_id
     # 根据执行模式进行对应修改
     mode = request.POST.get("mode")
@@ -424,7 +421,7 @@ def timing_task(request):
             # 增加工单日志
             audit_id = Audit.detail_by_workflow_id(
                 workflow_id=workflow_id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             ).audit_id
             Audit.add_log(
                 audit_id=audit_id,
@@ -468,7 +465,7 @@ def cancel(request):
             # 调用工作流接口取消或者驳回
             workflow_audit = Audit.detail_by_workflow_id(
                 workflow_id=workflow_id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             )
             audit_id = workflow_audit.audit_id
             # 仅待审核的需要调用工作流，审核通过的不需要
@@ -496,7 +493,7 @@ def cancel(request):
                 if user.username == sql_workflow.engineer:
                     _, workflow_audit_detail = Audit.audit(
                         audit_id,
-                        WorkflowDict.workflow_status["audit_abort"],
+                        WorkflowStatus.ABORTED,
                         user.username,
                         audit_remark,
                     )
@@ -504,7 +501,7 @@ def cancel(request):
                 elif user.has_perm("sql.sql_review"):
                     _, workflow_audit_detail = Audit.audit(
                         audit_id,
-                        WorkflowDict.workflow_status["audit_reject"],
+                        WorkflowStatus.REJECTED,
                         user.username,
                         audit_remark,
                     )
@@ -533,8 +530,8 @@ def cancel(request):
         if is_notified:
             workflow_audit.refresh_from_db()
             if workflow_audit.current_status in (
-                WorkflowDict.workflow_status["audit_abort"],
-                WorkflowDict.workflow_status["audit_reject"],
+                WorkflowStatus.ABORTED,
+                WorkflowStatus.REJECTED,
             ):
                 async_task(
                     notify_for_audit,
diff --git a/sql/tests.py b/sql/tests.py
index 8f3fd12fe4..0e5226480c 100644
--- a/sql/tests.py
+++ b/sql/tests.py
@@ -9,7 +9,7 @@
 
 import sql.query_privileges
 from common.config import SysConfig
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus
 from sql.archiver import add_archive_task, archive
 from sql.binlog import my2sql_file
 from sql.engines.models import ResultSet
@@ -1946,7 +1946,7 @@ def setUp(self):
             mode="file",
             no_delete=True,
             sleep=1,
-            status=WorkflowDict.workflow_status["audit_wait"],
+            status=WorkflowStatus.WAITING,
             state=False,
             user_name="some_user",
             user_display="display",
@@ -2108,7 +2108,7 @@ def test_archive_audit(self, _async_task, _audit):
         )
         data = {
             "archive_id": self.archive_apply.id,
-            "audit_status": WorkflowDict.workflow_status["audit_success"],
+            "audit_status": WorkflowStatus.PASSED,
             "audit_remark": "xxxx",
         }
         self.client.force_login(self.superuser)
diff --git a/sql/utils/execute_sql.py b/sql/utils/execute_sql.py
index cb53999739..8027a4c9aa 100644
--- a/sql/utils/execute_sql.py
+++ b/sql/utils/execute_sql.py
@@ -4,7 +4,7 @@
 
 from django.db import close_old_connections, connection, transaction
 from django_redis import get_redis_connection
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from common.config import SysConfig
 from sql.engines.models import ReviewResult, ReviewSet
 from sql.models import SqlWorkflow
@@ -30,7 +30,7 @@ def execute(workflow_id, user=None):
             )
     # 增加执行日志
     audit_id = Audit.detail_by_workflow_id(
-        workflow_id=workflow_id, workflow_type=WorkflowDict.workflow_type["sqlreview"]
+        workflow_id=workflow_id, workflow_type=WorkflowType.SQL_REVIEW
     ).audit_id
     Audit.add_log(
         audit_id=audit_id,
@@ -95,7 +95,7 @@ def execute_callback(task):
         workflow.sqlworkflowcontent.save()
     # 增加工单日志
     audit_id = Audit.detail_by_workflow_id(
-        workflow_id=workflow_id, workflow_type=WorkflowDict.workflow_type["sqlreview"]
+        workflow_id=workflow_id, workflow_type=WorkflowType.SQL_REVIEW
     ).audit_id
     Audit.add_log(
         audit_id=audit_id,
diff --git a/sql/utils/tests.py b/sql/utils/tests.py
index 1c69b9d207..4fa4edfebc 100644
--- a/sql/utils/tests.py
+++ b/sql/utils/tests.py
@@ -21,7 +21,7 @@
 from django_q.models import Schedule
 
 from common.config import SysConfig
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from sql.engines.models import ReviewResult, ReviewSet, SqlItem
 from sql.models import (
     Users,
@@ -1143,7 +1143,7 @@ def setUp(self):
             mode="file",
             no_delete=True,
             sleep=1,
-            status=WorkflowDict.workflow_status["audit_wait"],
+            status=WorkflowStatus.WAITING,
             state=False,
             user_name="some_user",
             user_display="display",
@@ -1182,7 +1182,7 @@ def test_audit_add_query(self):
         result, _ = Audit.add(1, self.query_apply_1.apply_id)
         audit_id = result["data"]["audit_id"]
         workflow_status = result["data"]["workflow_status"]
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_wait"])
+        self.assertEqual(workflow_status, WorkflowStatus.WAITING)
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
         # 当前审批
         self.assertEqual(audit_detail.current_audit, "some_audit_group")
@@ -1199,7 +1199,7 @@ def test_audit_add_sqlreview(self):
         result, _ = Audit.add(2, self.wf.id)
         audit_id = result["data"]["audit_id"]
         workflow_status = result["data"]["workflow_status"]
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_wait"])
+        self.assertEqual(workflow_status, WorkflowStatus.WAITING)
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
         # 当前审批
         self.assertEqual(audit_detail.current_audit, "some_audit_group")
@@ -1216,7 +1216,7 @@ def test_audit_add_archive_review(self):
         result, workflow_audit_detail = Audit.add(3, self.archive_apply_1.id)
         audit_id = result["data"]["audit_id"]
         workflow_status = result["data"]["workflow_status"]
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_wait"])
+        self.assertEqual(workflow_status, WorkflowStatus.WAITING)
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
         # 当前审批
         self.assertEqual(audit_detail.current_audit, "some_audit_group")
@@ -1253,7 +1253,7 @@ def test_audit_add_auto_review(self, _is_auto_review):
         result, workflow_audit_detail = Audit.add(2, self.wf.id)
         audit_id = result["data"]["audit_id"]
         workflow_status = result["data"]["workflow_status"]
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_success"])
+        self.assertEqual(workflow_status, WorkflowStatus.PASSED)
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
         # 无下级审批
         self.assertEqual(audit_detail.next_audit, "-1")
@@ -1271,7 +1271,7 @@ def test_audit_add_multiple_audit(self):
         audit_id = result["data"]["audit_id"]
         workflow_status = result["data"]["workflow_status"]
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_wait"])
+        self.assertEqual(workflow_status, WorkflowStatus.WAITING)
         # 存在下级审批
         self.assertEqual(audit_detail.current_audit, "1")
         self.assertEqual(audit_detail.next_audit, "2")
@@ -1288,14 +1288,14 @@ def test_audit_success_not_exists_next(self):
         self.audit.save()
         result, _ = Audit.audit(
             self.audit.audit_id,
-            WorkflowDict.workflow_status["audit_success"],
+            WorkflowStatus.PASSED,
             self.user.username,
             "通过",
         )
         audit_id = self.audit.audit_id
         workflow_status = result["data"]["workflow_status"]
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_success"])
+        self.assertEqual(workflow_status, WorkflowStatus.PASSED)
         # 不存在下级审批
         self.assertEqual(audit_detail.next_audit, "-1")
         # 验证日志
@@ -1313,14 +1313,14 @@ def test_audit_success_exists_next(self):
         self.audit.save()
         result, _ = Audit.audit(
             self.audit.audit_id,
-            WorkflowDict.workflow_status["audit_success"],
+            WorkflowStatus.PASSED,
             self.user.username,
             "通过",
         )
         audit_id = self.audit.audit_id
         workflow_status = result["data"]["workflow_status"]
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_wait"])
+        self.assertEqual(workflow_status, WorkflowStatus.WAITING)
         # 存在下级审批
         self.assertEqual(audit_detail.next_audit, "3")
         # 验证日志
@@ -1335,14 +1335,14 @@ def test_audit_reject(self):
         """测试审核不通过"""
         result, _ = Audit.audit(
             self.audit.audit_id,
-            WorkflowDict.workflow_status["audit_reject"],
+            WorkflowStatus.REJECTED,
             self.user.username,
             "不通过",
         )
         audit_id = self.audit.audit_id
         workflow_status = result["data"]["workflow_status"]
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_reject"])
+        self.assertEqual(workflow_status, WorkflowStatus.REJECTED)
         # 不存在下级审批
         self.assertEqual(audit_detail.next_audit, "-1")
         # 验证日志
@@ -1359,14 +1359,14 @@ def test_audit_abort(self):
         self.audit.save()
         result, _ = Audit.audit(
             self.audit.audit_id,
-            WorkflowDict.workflow_status["audit_abort"],
+            WorkflowStatus.ABORTED,
             self.user.username,
             "取消",
         )
         audit_id = self.audit.audit_id
         workflow_status = result["data"]["workflow_status"]
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
-        self.assertEqual(workflow_status, WorkflowDict.workflow_status["audit_abort"])
+        self.assertEqual(workflow_status, WorkflowStatus.ABORTED)
         # 不存在下级审批
         self.assertEqual(audit_detail.next_audit, "-1")
         # 验证日志
@@ -1389,7 +1389,7 @@ def test_audit_success_wrong_status(self):
         with self.assertRaisesMessage(Exception, "工单不是待审核状态，请返回刷新"):
             Audit.audit(
                 self.audit.audit_id,
-                WorkflowDict.workflow_status["audit_success"],
+                WorkflowStatus.PASSED,
                 self.user.username,
                 "",
             )
@@ -1401,7 +1401,7 @@ def test_audit_reject_wrong_status(self):
         with self.assertRaisesMessage(Exception, "工单不是待审核状态，请返回刷新"):
             Audit.audit(
                 self.audit.audit_id,
-                WorkflowDict.workflow_status["audit_reject"],
+                WorkflowStatus.REJECTED,
                 self.user.username,
                 "",
             )
@@ -1413,7 +1413,7 @@ def test_audit_abort_wrong_status(self):
         with self.assertRaisesMessage(Exception, "工单不是待审核态/审核通过状态，请返回刷新"):
             Audit.audit(
                 self.audit.audit_id,
-                WorkflowDict.workflow_status["audit_abort"],
+                WorkflowStatus.ABORTED,
                 self.user.username,
                 "",
             )
@@ -1433,12 +1433,10 @@ def test_detail(self):
 
     def test_detail_by_workflow_id(self):
         """测试通过业务id获取审核信息"""
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.save()
-        result = Audit.detail_by_workflow_id(
-            self.wf.id, WorkflowDict.workflow_type["sqlreview"]
-        )
+        result = Audit.detail_by_workflow_id(self.wf.id, WorkflowType.SQL_REVIEW)
         self.assertEqual(result, self.audit)
         result = Audit.detail_by_workflow_id(0, 0)
         self.assertEqual(result, None)
@@ -1477,7 +1475,7 @@ def test_can_review_sql_review(self, _detail_by_workflow_id, _auth_group_users):
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.save()
         r = Audit.can_review(
@@ -1497,7 +1495,7 @@ def test_cannot_review_self_sql_review(
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.own_wf.id
         self.audit.save()
         r = Audit.can_review(
@@ -1514,7 +1512,7 @@ def test_can_review_query_review(self, _detail_by_workflow_id, _auth_group_users
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["query"]
+        self.audit.workflow_type = WorkflowType.QUERY
         self.audit.workflow_id = self.query_apply_1.apply_id
         self.audit.save()
         r = Audit.can_review(
@@ -1531,7 +1529,7 @@ def test_can_review_sql_review_super(
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.save()
         r = Audit.can_review(self.su, self.audit.workflow_id, self.audit.workflow_type)
@@ -1544,9 +1542,9 @@ def test_can_review_wrong_status(self, _detail_by_workflow_id, _auth_group_users
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
-        self.audit.current_status = WorkflowDict.workflow_status["audit_success"]
+        self.audit.current_status = WorkflowStatus.PASSED
         self.audit.save()
         r = Audit.can_review(
             self.user, self.audit.workflow_id, self.audit.workflow_type
@@ -1560,7 +1558,7 @@ def test_can_review_no_prem(self, _detail_by_workflow_id, _auth_group_users):
         aug = Group.objects.create(name="auth_group")
         _detail_by_workflow_id.return_value.current_audit = aug.id
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.save()
         r = Audit.can_review(
@@ -1577,7 +1575,7 @@ def test_can_review_no_prem_exception(
         Group.objects.create(name="auth_group")
         _detail_by_workflow_id.side_effect = RuntimeError()
         _auth_group_users.return_value.filter.exists = True
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.save()
         with self.assertRaisesMessage(Exception, "当前审批auth_group_id不存在，请检查并清洗历史数据"):
@@ -1587,7 +1585,7 @@ def test_can_review_no_prem_exception(
 
     def test_review_info_no_review(self):
         """测试获取当前工单审批流程和当前审核组，无需审批"""
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.audit_auth_groups = ""
         self.audit.current_audit = "-1"
@@ -1601,7 +1599,7 @@ def test_review_info_no_review(self):
     def test_review_info(self):
         """测试获取当前工单审批流程和当前审核组，无需审批"""
         aug = Group.objects.create(name="DBA")
-        self.audit.workflow_type = WorkflowDict.workflow_type["sqlreview"]
+        self.audit.workflow_type = WorkflowType.SQL_REVIEW
         self.audit.workflow_id = self.wf.id
         self.audit.audit_auth_groups = str(aug.id)
         self.audit.current_audit = str(aug.id)
diff --git a/sql/utils/workflow_audit.py b/sql/utils/workflow_audit.py
index ae64382d09..266cacbfa6 100644
--- a/sql/utils/workflow_audit.py
+++ b/sql/utils/workflow_audit.py
@@ -4,7 +4,7 @@
 
 from sql.utils.resource_group import user_groups, auth_group_users
 from sql.utils.sql_review import is_auto_review
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowStatus, WorkflowType
 from sql.models import (
     WorkflowAudit,
     WorkflowAuditDetail,
@@ -29,14 +29,14 @@ def add(workflow_type, workflow_id):
         workflow_info = WorkflowAudit.objects.filter(
             workflow_type=workflow_type,
             workflow_id=workflow_id,
-            current_status=WorkflowDict.workflow_status["audit_wait"],
+            current_status=WorkflowStatus.WAITING,
         )
         if len(workflow_info) >= 1:
             result["msg"] = "该工单当前状态为待审核，请勿重复提交"
             raise Exception(result["msg"])
 
         # 获取工单信息
-        if workflow_type == WorkflowDict.workflow_type["query"]:
+        if workflow_type == WorkflowType.QUERY:
             workflow_detail = QueryPrivilegesApply.objects.get(apply_id=workflow_id)
             workflow_title = workflow_detail.title
             group_id = workflow_detail.group_id
@@ -45,7 +45,7 @@ def add(workflow_type, workflow_id):
             create_user_display = workflow_detail.user_display
             audit_auth_groups = workflow_detail.audit_auth_groups
             workflow_remark = ""
-        elif workflow_type == WorkflowDict.workflow_type["sqlreview"]:
+        elif workflow_type == WorkflowType.SQL_REVIEW:
             workflow_detail = SqlWorkflow.objects.get(pk=workflow_id)
             workflow_title = workflow_detail.workflow_name
             group_id = workflow_detail.group_id
@@ -54,7 +54,7 @@ def add(workflow_type, workflow_id):
             create_user_display = workflow_detail.engineer_display
             audit_auth_groups = workflow_detail.audit_auth_groups
             workflow_remark = ""
-        elif workflow_type == WorkflowDict.workflow_type["archive"]:
+        elif workflow_type == WorkflowType.ARCHIVE:
             workflow_detail = ArchiveConfig.objects.get(pk=workflow_id)
             workflow_title = workflow_detail.title
             group_id = workflow_detail.resource_group.group_id
@@ -76,7 +76,7 @@ def add(workflow_type, workflow_id):
 
         # 判断是否无需审核,并且修改审批人为空
         if SysConfig().get("auto_review", False):
-            if workflow_type == WorkflowDict.workflow_type["sqlreview"]:
+            if workflow_type == WorkflowType.SQL_REVIEW:
                 if is_auto_review(workflow_id):
                     sql_workflow = SqlWorkflow.objects.get(id=int(workflow_id))
                     sql_workflow.audit_auth_groups = "无需审批"
@@ -97,15 +97,11 @@ def add(workflow_type, workflow_id):
             audit_detail.audit_auth_groups = ""
             audit_detail.current_audit = "-1"
             audit_detail.next_audit = "-1"
-            audit_detail.current_status = WorkflowDict.workflow_status[
-                "audit_success"
-            ]  # 审核通过
+            audit_detail.current_status = WorkflowStatus.PASSED  # 审核通过
             audit_detail.create_user = create_user
             audit_detail.create_user_display = create_user_display
             audit_detail.save()
-            result["data"] = {
-                "workflow_status": WorkflowDict.workflow_status["audit_success"]
-            }
+            result["data"] = {"workflow_status": WorkflowStatus.PASSED}
             result["msg"] = "无审核配置，直接审核通过"
             # 增加工单日志
             Audit.add_log(
@@ -133,13 +129,11 @@ def add(workflow_type, workflow_id):
             else:
                 audit_detail.next_audit = audit_auth_groups_list[1]
 
-            audit_detail.current_status = WorkflowDict.workflow_status["audit_wait"]
+            audit_detail.current_status = WorkflowStatus.WAITING
             audit_detail.create_user = create_user
             audit_detail.create_user_display = create_user_display
             audit_detail.save()
-            result["data"] = {
-                "workflow_status": WorkflowDict.workflow_status["audit_wait"]
-            }
+            result["data"] = {"workflow_status": WorkflowStatus.WAITING}
             # 增加工单日志
             audit_auth_group, current_audit_auth_group = Audit.review_info(
                 workflow_id, workflow_type
@@ -166,12 +160,9 @@ def audit(
         audit_detail = WorkflowAudit.objects.get(audit_id=audit_id)
 
         # 不同审核状态
-        if audit_status == WorkflowDict.workflow_status["audit_success"]:
+        if audit_status == WorkflowStatus.PASSED:
             # 判断当前工单是否为待审核状态
-            if (
-                audit_detail.current_status
-                != WorkflowDict.workflow_status["audit_wait"]
-            ):
+            if audit_detail.current_status != WorkflowStatus.WAITING:
                 result["msg"] = "工单不是待审核状态，请返回刷新"
                 raise Exception(result["msg"])
 
@@ -181,15 +172,13 @@ def audit(
                 audit_result = WorkflowAudit()
                 audit_result.audit_id = audit_id
                 audit_result.current_audit = "-1"
-                audit_result.current_status = WorkflowDict.workflow_status[
-                    "audit_success"
-                ]
+                audit_result.current_status = WorkflowStatus.PASSED
                 audit_result.save(update_fields=["current_audit", "current_status"])
             else:
                 # 更新主表审核下级审核组和当前审核组
                 audit_result = WorkflowAudit()
                 audit_result.audit_id = audit_id
-                audit_result.current_status = WorkflowDict.workflow_status["audit_wait"]
+                audit_result.current_status = WorkflowStatus.WAITING
                 audit_result.current_audit = audit_detail.next_audit
                 # 判断后续是否还有下下一级审核组
                 audit_auth_groups_list = audit_detail.audit_auth_groups.split(",")
@@ -210,9 +199,7 @@ def audit(
             audit_detail_result = WorkflowAuditDetail()
             audit_detail_result.audit_id = audit_id
             audit_detail_result.audit_user = audit_user
-            audit_detail_result.audit_status = WorkflowDict.workflow_status[
-                "audit_success"
-            ]
+            audit_detail_result.audit_status = WorkflowStatus.PASSED
             audit_detail_result.audit_time = timezone.now()
             audit_detail_result.remark = audit_remark
             audit_detail_result.save()
@@ -230,12 +217,9 @@ def audit(
                 operator=audit_user,
                 operator_display=Users.objects.get(username=audit_user).display,
             )
-        elif audit_status == WorkflowDict.workflow_status["audit_reject"]:
+        elif audit_status == WorkflowStatus.REJECTED:
             # 判断当前工单是否为待审核状态
-            if (
-                audit_detail.current_status
-                != WorkflowDict.workflow_status["audit_wait"]
-            ):
+            if audit_detail.current_status != WorkflowStatus.WAITING:
                 result["msg"] = "工单不是待审核状态，请返回刷新"
                 raise Exception(result["msg"])
 
@@ -244,7 +228,7 @@ def audit(
             audit_result.audit_id = audit_id
             audit_result.current_audit = "-1"
             audit_result.next_audit = "-1"
-            audit_result.current_status = WorkflowDict.workflow_status["audit_reject"]
+            audit_result.current_status = WorkflowStatus.REJECTED
             audit_result.save(
                 update_fields=["current_audit", "next_audit", "current_status"]
             )
@@ -253,9 +237,7 @@ def audit(
             audit_detail_result = WorkflowAuditDetail()
             audit_detail_result.audit_id = audit_id
             audit_detail_result.audit_user = audit_user
-            audit_detail_result.audit_status = WorkflowDict.workflow_status[
-                "audit_reject"
-            ]
+            audit_detail_result.audit_status = WorkflowStatus.REJECTED
             audit_detail_result.audit_time = timezone.now()
             audit_detail_result.remark = audit_remark
             audit_detail_result.save()
@@ -268,13 +250,11 @@ def audit(
                 operator=audit_user,
                 operator_display=Users.objects.get(username=audit_user).display,
             )
-        elif audit_status == WorkflowDict.workflow_status["audit_abort"]:
+        elif audit_status == WorkflowStatus.ABORTED:
             # 判断当前工单是否为待审核/审核通过状态
             if (
-                audit_detail.current_status
-                != WorkflowDict.workflow_status["audit_wait"]
-                and audit_detail.current_status
-                != WorkflowDict.workflow_status["audit_success"]
+                audit_detail.current_status != WorkflowStatus.WAITING
+                and audit_detail.current_status != WorkflowStatus.PASSED
             ):
                 result["msg"] = "工单不是待审核态/审核通过状态，请返回刷新"
                 raise Exception(result["msg"])
@@ -283,16 +263,14 @@ def audit(
             audit_result = WorkflowAudit()
             audit_result.audit_id = audit_id
             audit_result.next_audit = "-1"
-            audit_result.current_status = WorkflowDict.workflow_status["audit_abort"]
+            audit_result.current_status = WorkflowStatus.ABORTED
             audit_result.save(update_fields=["current_status", "next_audit"])
 
             # 插入审核明细数据
             audit_detail_result = WorkflowAuditDetail()
             audit_detail_result.audit_id = audit_id
             audit_detail_result.audit_user = audit_user
-            audit_detail_result.audit_status = WorkflowDict.workflow_status[
-                "audit_abort"
-            ]
+            audit_detail_result.audit_status = WorkflowStatus.ABORTED
             audit_detail_result.audit_time = timezone.now()
             audit_detail_result.remark = audit_remark
             audit_detail_result.save()
@@ -327,7 +305,7 @@ def todo(user):
             auth_group_ids = [group.id for group in Group.objects.filter(user=user)]
 
         return WorkflowAudit.objects.filter(
-            current_status=WorkflowDict.workflow_status["audit_wait"],
+            current_status=WorkflowStatus.WAITING,
             group_id__in=group_ids,
             current_audit__in=auth_group_ids,
         ).count()
@@ -408,7 +386,7 @@ def get_workflow_applicant(workflow_id, workflow_type):
         ):
             return result
         # 只有待审核状态数据才可以审核
-        if audit_info.current_status == WorkflowDict.workflow_status["audit_wait"]:
+        if audit_info.current_status == WorkflowStatus.WAITING:
             try:
                 auth_group_id = Audit.detail_by_workflow_id(
                     workflow_id, workflow_type
diff --git a/sql/views.py b/sql/views.py
index bbc986f074..91fad2f83e 100644
--- a/sql/views.py
+++ b/sql/views.py
@@ -40,7 +40,7 @@
     can_view,
     can_rollback,
 )
-from common.utils.const import Const, WorkflowDict
+from common.utils.const import Const, WorkflowType
 from sql.utils.resource_group import user_groups, user_instances, auth_group_users
 
 import logging
@@ -207,7 +207,7 @@ def detail(request, workflow_id):
         try:
             audit_detail = Audit.detail_by_workflow_id(
                 workflow_id=workflow_id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             )
             audit_id = audit_detail.audit_id
             last_operation_info = (
@@ -528,7 +528,7 @@ def config(request):
         "instance_tags": instance_tags,
         "db_type": db_type,
         "config": sys_config,
-        "WorkflowDict": WorkflowDict,
+        "workflow_choices": WorkflowType,
     }
     return render(request, "config.html", context)
 
@@ -557,15 +557,15 @@ def workflowsdetail(request, audit_id):
     audit_detail = Audit.detail(audit_id)
     if not audit_detail:
         raise Http404("不存在对应的工单记录")
-    if audit_detail.workflow_type == WorkflowDict.workflow_type["query"]:
+    if audit_detail.workflow_type == WorkflowType.QUERY:
         return HttpResponseRedirect(
             reverse("sql:queryapplydetail", args=(audit_detail.workflow_id,))
         )
-    elif audit_detail.workflow_type == WorkflowDict.workflow_type["sqlreview"]:
+    elif audit_detail.workflow_type == WorkflowType.SQL_REVIEW:
         return HttpResponseRedirect(
             reverse("sql:detail", args=(audit_detail.workflow_id,))
         )
-    elif audit_detail.workflow_type == WorkflowDict.workflow_type["archive"]:
+    elif audit_detail.workflow_type == WorkflowType.ARCHIVE:
         return HttpResponseRedirect(
             reverse("sql:archive_detail", args=(audit_detail.workflow_id,))
         )
diff --git a/sql_api/api_workflow.py b/sql_api/api_workflow.py
index c379750777..e8e0e3b785 100644
--- a/sql_api/api_workflow.py
+++ b/sql_api/api_workflow.py
@@ -1,8 +1,35 @@
+import datetime
+import logging
+import traceback
+
 from django.contrib.auth.decorators import permission_required
+from django.contrib.auth.models import Group
+from django.db import transaction
 from django.utils.decorators import method_decorator
+from django_q.tasks import async_task
+from drf_spectacular.utils import extend_schema
 from rest_framework import views, generics, status, serializers, permissions
 from rest_framework.response import Response
-from drf_spectacular.utils import extend_schema
+
+from common.config import SysConfig
+from common.utils.const import WorkflowStatus, WorkflowType
+from sql.engines import get_engine
+from sql.models import (
+    SqlWorkflow,
+    SqlWorkflowContent,
+    WorkflowAudit,
+    Users,
+    WorkflowLog,
+    ArchiveConfig,
+)
+from sql.notify import notify_for_audit, notify_for_execute
+from sql.query_privileges import _query_apply_audit_call_back
+from sql.utils.resource_group import user_groups
+from sql.utils.sql_review import can_cancel, can_execute, on_correct_time_period
+from sql.utils.tasks import del_schedule
+from sql.utils.workflow_audit import Audit
+from .filters import WorkflowFilter, WorkflowAuditFilter
+from .pagination import CustomizedPagination
 from .serializers import (
     WorkflowContentSerializer,
     ExecuteCheckSerializer,
@@ -14,31 +41,6 @@
     AuditWorkflowSerializer,
     ExecuteWorkflowSerializer,
 )
-from .pagination import CustomizedPagination
-from .filters import WorkflowFilter, WorkflowAuditFilter
-from sql.models import (
-    SqlWorkflow,
-    SqlWorkflowContent,
-    WorkflowAudit,
-    Users,
-    WorkflowLog,
-    ArchiveConfig,
-)
-from sql.utils.sql_review import can_cancel, can_execute, on_correct_time_period
-from sql.utils.resource_group import user_groups
-from sql.utils.workflow_audit import Audit
-from sql.utils.tasks import del_schedule
-from sql.notify import notify_for_audit, notify_for_execute
-from sql.query_privileges import _query_apply_audit_call_back
-from sql.engines import get_engine
-from common.utils.const import WorkflowDict
-from common.config import SysConfig
-from django.contrib.auth.models import Group
-from django.db import transaction
-from django_q.tasks import async_task
-import traceback
-import datetime
-import logging
 
 logger = logging.getLogger("default")
 
@@ -144,7 +146,7 @@ def post(self, request):
             # 获取审核信息
             workflow_audit = Audit.detail_by_workflow_id(
                 workflow_id=workflow_content.workflow.id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             )
             async_task(
                 notify_for_audit,
@@ -164,7 +166,7 @@ class WorkflowAuditList(generics.ListAPIView):
     pagination_class = CustomizedPagination
     serializer_class = WorkflowAuditListSerializer
     queryset = WorkflowAudit.objects.filter(
-        current_status=WorkflowDict.workflow_status["audit_wait"]
+        current_status=WorkflowStatus.WAITING
     ).order_by("-audit_id")
 
     @extend_schema(exclude=True)
@@ -195,7 +197,7 @@ def post(self, request):
             auth_group_ids = [group.id for group in Group.objects.filter(user=user)]
 
         self.queryset = self.queryset.filter(
-            current_status=WorkflowDict.workflow_status["audit_wait"],
+            current_status=WorkflowStatus.WAITING,
             group_id__in=group_ids,
             current_audit__in=auth_group_ids,
         )
@@ -242,7 +244,7 @@ def post(self, request):
                 with transaction.atomic():
                     workflow_audit = Audit.detail_by_workflow_id(
                         workflow_id=workflow_id,
-                        workflow_type=WorkflowDict.workflow_type["query"],
+                        workflow_type=WorkflowType.QUERY,
                     )
                     audit_id = workflow_audit.audit_id
 
@@ -253,10 +255,7 @@ def post(self, request):
 
                     # 按照审核结果更新业务表审核状态
                     workflow_audit = Audit.detail(audit_id)
-                    if (
-                        workflow_audit.workflow_type
-                        == WorkflowDict.workflow_type["query"]
-                    ):
+                    if workflow_audit.workflow_type == WorkflowType.QUERY:
                         # 更新业务表审核状态,插入权限信息
                         _query_apply_audit_call_back(
                             workflow_audit.workflow_id,
@@ -294,12 +293,12 @@ def post(self, request):
                         # 调用工作流接口审核
                         workflow_audit = Audit.detail_by_workflow_id(
                             workflow_id=workflow_id,
-                            workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                            workflow_type=WorkflowType.SQL_REVIEW,
                         )
                         audit_id = workflow_audit.audit_id
                         audit_result, audit_detail = Audit.audit(
                             audit_id,
-                            WorkflowDict.workflow_status["audit_success"],
+                            WorkflowStatus.PASSED,
                             user.username,
                             audit_remark,
                         )
@@ -307,7 +306,7 @@ def post(self, request):
                         # 按照审核结果更新业务表审核状态
                         if (
                             audit_result["data"]["workflow_status"]
-                            == WorkflowDict.workflow_status["audit_success"]
+                            == WorkflowStatus.PASSED
                         ):
                             # 将流程状态修改为审核通过
                             SqlWorkflow(
@@ -349,7 +348,7 @@ def post(self, request):
                         # 调用工作流接口取消或者驳回
                         audit_id = Audit.detail_by_workflow_id(
                             workflow_id=workflow_id,
-                            workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                            workflow_type=WorkflowType.SQL_REVIEW,
                         ).audit_id
                         # 仅待审核的需要调用工作流，审核通过的不需要
                         if workflow_detail.status != "workflow_manreviewing":
@@ -376,7 +375,7 @@ def post(self, request):
                             if user.username == workflow_detail.engineer:
                                 _, audit_detail = Audit.audit(
                                     audit_id,
-                                    WorkflowDict.workflow_status["audit_abort"],
+                                    WorkflowStatus.ABORTED,
                                     user.username,
                                     audit_remark,
                                 )
@@ -384,7 +383,7 @@ def post(self, request):
                             elif user.has_perm("sql.sql_review"):
                                 _, audit_detail = Audit.audit(
                                     audit_id,
-                                    WorkflowDict.workflow_status["audit_reject"],
+                                    WorkflowStatus.REJECTED,
                                     user.username,
                                     audit_remark,
                                 )
@@ -414,11 +413,11 @@ def post(self, request):
                     if is_notified:
                         workflow_audit = Audit.detail_by_workflow_id(
                             workflow_id=workflow_id,
-                            workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                            workflow_type=WorkflowType.SQL_REVIEW,
                         )
                         if workflow_audit.current_status in (
-                            WorkflowDict.workflow_status["audit_abort"],
-                            WorkflowDict.workflow_status["audit_reject"],
+                            WorkflowStatus.ABORTED,
+                            WorkflowStatus.REJECTED,
                         ):
                             async_task(
                                 notify_for_audit,
@@ -443,7 +442,7 @@ def post(self, request):
                 with transaction.atomic():
                     workflow_audit = Audit.detail_by_workflow_id(
                         workflow_id=workflow_id,
-                        workflow_type=WorkflowDict.workflow_type["archive"],
+                        workflow_type=WorkflowType.ARCHIVE,
                     )
                     audit_id = workflow_audit.audit_id
 
@@ -455,9 +454,7 @@ def post(self, request):
                     ArchiveConfig(
                         id=workflow_id,
                         status=audit_status,
-                        state=True
-                        if audit_status == WorkflowDict.workflow_status["audit_success"]
-                        else False,
+                        state=True if audit_status == WorkflowStatus.PASSED else False,
                     ).save(update_fields=["status", "state"])
             except Exception as msg:
                 logger.error(traceback.format_exc())
@@ -519,7 +516,7 @@ def post(self, request):
             # 获取审核信息
             audit_id = Audit.detail_by_workflow_id(
                 workflow_id=workflow_id,
-                workflow_type=WorkflowDict.workflow_type["sqlreview"],
+                workflow_type=WorkflowType.SQL_REVIEW,
             ).audit_id
 
             # 交由系统执行
diff --git a/sql_api/serializers.py b/sql_api/serializers.py
index 7bfb5156c4..389e00809c 100644
--- a/sql_api/serializers.py
+++ b/sql_api/serializers.py
@@ -20,7 +20,7 @@
 from sql.engines import get_engine
 from sql.utils.workflow_audit import Audit
 from sql.utils.resource_group import user_instances
-from common.utils.const import WorkflowDict
+from common.utils.const import WorkflowType
 from common.config import SysConfig
 import traceback
 import logging
@@ -414,7 +414,7 @@ def create(self, validated_data):
             engineer_display=user.display,
             group_name=group.group_name,
             audit_auth_groups=Audit.settings(
-                workflow_data["group_id"], WorkflowDict.workflow_type["sqlreview"]
+                workflow_data["group_id"], WorkflowType.SQL_REVIEW
             ),
         )
         try:
@@ -427,7 +427,7 @@ def create(self, validated_data):
                 # 自动审核通过了，才调用工作流
                 if workflow_status == "workflow_manreviewing":
                     # 调用工作流插入审核信息, SQL上线权限申请workflow_type=2
-                    Audit.add(WorkflowDict.workflow_type["sqlreview"], workflow.id)
+                    Audit.add(WorkflowType.SQL_REVIEW, workflow.id)
         except Exception as e:
             logger.error(f"提交工单报错，错误信息：{traceback.format_exc()}")
             raise serializers.ValidationError({"errors": str(e)})