diff --git a/archery/settings.py b/archery/settings.py
index 7780d26b49..c01e6bb1b6 100644
--- a/archery/settings.py
+++ b/archery/settings.py
@@ -3,6 +3,7 @@
 
 # Build paths inside the project like this: os.path.join(BASE_DIR, ...)
 import os
+from datetime import timedelta
 
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
@@ -32,6 +33,9 @@
     'sql',
     'sql_api',
     'common',
+    'rest_framework',
+    'django_filters',
+    'drf_spectacular',
 )
 
 MIDDLEWARE = (
@@ -172,6 +176,49 @@
     }
 }
 
+# API Framework
+REST_FRAMEWORK = {
+    'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
+    'DEFAULT_RENDERER_CLASSES': ('rest_framework.renderers.JSONRenderer',),
+    # 鉴权
+    'DEFAULT_AUTHENTICATION_CLASSES': (
+        'rest_framework_simplejwt.authentication.JWTAuthentication',
+        'rest_framework.authentication.SessionAuthentication',
+    ),
+    # 权限
+    'DEFAULT_PERMISSION_CLASSES': ('sql_api.permissions.IsInUserWhitelist',),
+    # 限速（anon：未认证用户  user：认证用户）
+    'DEFAULT_THROTTLE_CLASSES': (
+        'rest_framework.throttling.AnonRateThrottle',
+        'rest_framework.throttling.UserRateThrottle',
+    ),
+    'DEFAULT_THROTTLE_RATES': {
+        'anon': '120/min',
+        'user': '600/min'
+    },
+    # 过滤
+    'DEFAULT_FILTER_BACKENDS': ('django_filters.rest_framework.DjangoFilterBackend',),
+    # 分页
+    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
+    'PAGE_SIZE': 5,
+}
+
+# Swagger UI
+SPECTACULAR_SETTINGS = {
+    'TITLE': 'Archery API',
+    'DESCRIPTION': 'OpenAPI 3.0',
+    'VERSION': '1.0.0',
+}
+
+# API Authentication
+SIMPLE_JWT = {
+    'ACCESS_TOKEN_LIFETIME': timedelta(hours=4),
+    'REFRESH_TOKEN_LIFETIME': timedelta(days=3),
+    'ALGORITHM': 'HS256',
+    'SIGNING_KEY': SECRET_KEY,
+    'AUTH_HEADER_TYPES': ('Bearer',),
+}
+
 # LDAP
 ENABLE_LDAP = False
 if ENABLE_LDAP:
diff --git a/common/middleware/check_login_middleware.py b/common/middleware/check_login_middleware.py
index 3cf7bfbc6c..3707169e73 100644
--- a/common/middleware/check_login_middleware.py
+++ b/common/middleware/check_login_middleware.py
@@ -10,7 +10,7 @@
     '/api/info'
 ]
 
-IGNORE_URL_RE = r'/admin/\w*'
+IGNORE_URL_RE = r'(/admin/\w*|/api/(v1|auth)/\w+)'
 
 
 class CheckLoginMiddleware(MiddlewareMixin):
diff --git a/common/templates/base.html b/common/templates/base.html
index b991602fce..0b37ef1288 100644
--- a/common/templates/base.html
+++ b/common/templates/base.html
@@ -267,6 +267,11 @@
                             <a href="/dbaprinciples/"><i class="fa fa-sitemap fa-fw"></i> 相关文档</a>
                         </li>
                     {% endif %}
+                    {% if perms.sql.menu_openapi %}
+                        <li>
+                            <a href="/api/swagger/" target="_blank"><i class="fa fa-code-fork fa-fw"></i> OpenAPI</a>
+                        </li>
+                    {% endif %}
                 </ul>
             </div>
             <!-- /.sidebar-collapse -->
diff --git a/common/templates/config.html b/common/templates/config.html
index e1d612fe41..2fa27a38c0 100755
--- a/common/templates/config.html
+++ b/common/templates/config.html
@@ -809,6 +809,19 @@ <h4 style="color: darkgrey"><b>其他配置</b></h4>
                                     </select>
                                 </div>
                             </div>
+                            <div class="form-group">
+                                <label for="api_user_whitelist"
+                                       class="col-sm-4 control-label">API_USER_WHITELIST</label>
+                                <div class="col-sm-5">
+                                    <select id="api_user_whitelist" key="api_user_whitelist"
+                                            class="selectpicker show-tick form-control bs-select-hidden"
+                                            data-live-search="true"
+                                            data-none-selected-text="API用户白名单"
+                                            multiple data-selected-text-format="count > 5"
+                                            required>
+                                    </select>
+                                </div>
+                            </div>
                             <div class="form-group">
                                 <label for="lock_time_threshold"
                                        class="col-sm-4 control-label">LOCK_TIME_THRESHOLD</label>
@@ -903,8 +916,9 @@ <h5 class="control-label text-bold">当前审批流程：<b id="workflow_auditor
     <link href="{% static 'bootstrap-switch/css/bootstrap-switch.min.css' %}" rel="stylesheet" type="text/css"/>
     <script src="{% static 'bootstrap-switch/js/bootstrap-switch.min.js' %}"></script>
     <script>
-        // notify_phase_control参数处理
+
         $(document).ready(function (){
+            // notify_phase_control参数处理
             let cfg_value = "{{ config.notify_phase_control }}";
             let phases = ['Apply', 'Pass', 'Execute', 'Cancel'];
             // notify_phase_control未设置时默认全部开启
@@ -927,6 +941,36 @@ <h5 class="control-label text-bold">当前审批流程：<b id="workflow_auditor
             }
             $("#notify_phase_control").selectpicker('render');
             $("#notify_phase_control").selectpicker('refresh');
+
+            // api_user_whitelist参数处理
+            let api_config = "{{ config.api_user_whitelist }}";
+            $.ajax({
+                type: "post",
+                url: "/user/list/",
+                dataType: "json",
+                success: function (data) {
+                    if (data.status === 0) {
+                        let users = data.data;
+                        let user_arr = api_config.split(',');
+                        for (let i=0;i<users.length;i++) {
+                            let user = "";
+                            if (user_arr.indexOf(users[i].id.toString()) >= 0) {
+                                user = "<option value=\"" + users[i].id+ "\" selected>" + users[i].display + "(" + users[i].username + ")</option>"
+                            } else {
+                                user = "<option value=\"" + users[i].id+ "\">" + users[i].display + "(" + users[i].username + ")</option>"
+                            }
+                            $("#api_user_whitelist").append(user)
+                            $("#api_user_whitelist").selectpicker('render');
+                            $("#api_user_whitelist").selectpicker('refresh');
+                        }
+                    } else {
+                        alert(data.msg);
+                    }
+                },
+                error: function (XMLHttpRequest, textStatus, errorThrown) {
+                    alert(errorThrown);
+                }
+            });
         })
 
         //配置项切换
diff --git a/requirements.txt b/requirements.txt
index de2b9c1069..1c13740082 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -29,3 +29,8 @@ pycryptodome==3.10.1
 pandas==1.1.5
 pyodps==0.10.7.1
 clickhouse-driver==0.2.3
+djangorestframework==3.12.4
+djangorestframework-simplejwt==4.3.0
+django-filter==21.1
+drf-spectacular==0.22.0
+pyjwt==1.7.1
diff --git a/sql/fixtures/auth_group.sql b/sql/fixtures/auth_group.sql
index 25e33a1a21..679c59da60 100644
--- a/sql/fixtures/auth_group.sql
+++ b/sql/fixtures/auth_group.sql
@@ -23,7 +23,7 @@ where codename in ('menu_dashboard','menu_sqlcheck','menu_sqlworkflow','menu_sql
 insert into auth_group_permissions (group_id, permission_id)
 select 3,id
 from auth_permission
-where codename in ('menu_dashboard','menu_sqlcheck','menu_sqlworkflow','menu_sqlanalyze','menu_query','menu_sqlquery','menu_queryapplylist','menu_sqloptimize','menu_sqladvisor','menu_slowquery','menu_instance','menu_instance_list','menu_dbdiagnostic','menu_database','menu_instance_account','menu_param','menu_data_dictionary','menu_tools','menu_archive','menu_binlog2sql','menu_my2sql','menu_schemasync','menu_system','menu_document','sql_submit','sql_review','sql_execute_for_resource_group','sql_execute','sql_analyze','optimize_sqladvisor','optimize_sqltuning','optimize_soar','query_applypriv','query_mgtpriv','query_review','query_submit','query_all_instances','query_resource_group_instance','process_view','process_kill','tablespace_view','trx_view','trxandlocks_view','instance_account_manage','param_view','param_edit','data_dictionary_export','archive_apply','archive_review','archive_mgt');
+where codename in ('menu_dashboard','menu_sqlcheck','menu_sqlworkflow','menu_sqlanalyze','menu_query','menu_sqlquery','menu_queryapplylist','menu_sqloptimize','menu_sqladvisor','menu_slowquery','menu_instance','menu_instance_list','menu_dbdiagnostic','menu_database','menu_instance_account','menu_param','menu_data_dictionary','menu_tools','menu_archive','menu_binlog2sql','menu_my2sql','menu_schemasync','menu_system','menu_document','menu_openapi','sql_submit','sql_review','sql_execute_for_resource_group','sql_execute','sql_analyze','optimize_sqladvisor','optimize_sqltuning','optimize_soar','query_applypriv','query_mgtpriv','query_review','query_submit','query_all_instances','query_resource_group_instance','process_view','process_kill','tablespace_view','trx_view','trxandlocks_view','instance_account_manage','param_view','param_edit','data_dictionary_export','archive_apply','archive_review','archive_mgt');
 
 -- PM
 insert into auth_group_permissions (group_id, permission_id)
diff --git a/sql/models.py b/sql/models.py
index 3fa06286bc..bb4c87dfd3 100755
--- a/sql/models.py
+++ b/sql/models.py
@@ -705,6 +705,7 @@ class Meta:
             ('menu_schemasync', '菜单 SchemaSync'),
             ('menu_system', '菜单 系统管理'),
             ('menu_document', '菜单 相关文档'),
+            ('menu_openapi', '菜单 OpenAPI'),
             ('sql_submit', '提交SQL上线工单'),
             ('sql_review', '审核SQL上线工单'),
             ('sql_execute_for_resource_group', '执行SQL上线工单(资源组粒度)'),
diff --git a/sql/urls.py b/sql/urls.py
index 237695e6f2..03942c24a5 100644
--- a/sql/urls.py
+++ b/sql/urls.py
@@ -8,7 +8,7 @@
 import sql.sql_optimize
 from common import auth, config, workflow, dashboard, check
 from sql import views, sql_workflow, sql_analyze, query, slowlog, instance, instance_account, db_diagnostic, \
-    resource_group, binlog, data_dictionary, archiver, audit_log
+    resource_group, binlog, data_dictionary, archiver, audit_log, user
 from sql.utils import tasks
 from common.utils import ding_api
 
@@ -160,4 +160,5 @@
 
     path('audit/log/', audit_log.audit_log),
     path('audit/input/', audit_log.audit_input),
+    path('user/list/', user.lists),
 ]
diff --git a/sql/user.py b/sql/user.py
new file mode 100644
index 0000000000..8dcfb52d06
--- /dev/null
+++ b/sql/user.py
@@ -0,0 +1,19 @@
+# -*- coding: UTF-8 -*-
+import simplejson as json
+from common.utils.permission import superuser_required
+from common.utils.extend_json_encoder import ExtendJSONEncoder
+from django.http import HttpResponse
+from .models import Users
+
+
+@superuser_required
+def lists(request):
+    """获取用户列表"""
+    users = Users.objects.order_by('username')
+    users = users.values("id", "username", "display", "is_superuser", "is_staff", "is_active", "email")
+
+    rows = [row for row in users]
+
+    result = {"status": 0, "msg": "ok", "data": rows}
+    return HttpResponse(json.dumps(result, cls=ExtendJSONEncoder, bigint_as_string=True),
+                        content_type='application/json')
diff --git a/sql_api/api_instance.py b/sql_api/api_instance.py
new file mode 100644
index 0000000000..0229c19483
--- /dev/null
+++ b/sql_api/api_instance.py
@@ -0,0 +1,194 @@
+from rest_framework import views, generics, status, serializers
+from rest_framework.response import Response
+from drf_spectacular.utils import extend_schema
+from .serializers import InstanceSerializer, InstanceDetailSerializer, TunnelSerializer, \
+    AliyunRdsSerializer, InstanceResourceSerializer, InstanceResourceListSerializer
+from .pagination import CustomizedPagination
+from .filters import InstanceFilter
+from sql.models import Instance, Tunnel, AliyunRdsConfig
+from sql.engines import get_engine
+from django.http import Http404
+import MySQLdb
+
+
+class InstanceList(generics.ListAPIView):
+    """
+    列出所有的instance或者创建一个新的instance配置
+    """
+    filterset_class = InstanceFilter
+    pagination_class = CustomizedPagination
+    serializer_class = InstanceSerializer
+    queryset = Instance.objects.all().order_by('id')
+
+    @extend_schema(summary="实例清单",
+                   request=InstanceSerializer,
+                   responses={200: InstanceSerializer},
+                   description="列出所有实例（过滤，分页）")
+    def get(self, request):
+        instances = self.filter_queryset(self.queryset)
+        page_ins = self.paginate_queryset(queryset=instances)
+        serializer_obj = self.get_serializer(page_ins, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建实例",
+                   request=InstanceSerializer,
+                   responses={201: InstanceSerializer},
+                   description="创建一个实例配置")
+    def post(self, request):
+        serializer = InstanceSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class InstanceDetail(views.APIView):
+    """
+    实例操作
+    """
+    serializer_class = InstanceDetailSerializer
+
+    def get_object(self, pk):
+        try:
+            return Instance.objects.get(pk=pk)
+        except Instance.DoesNotExist:
+            raise Http404
+
+    @extend_schema(summary="更新实例",
+                   request=InstanceDetailSerializer,
+                   responses={200: InstanceDetailSerializer},
+                   description="更新一个实例配置")
+    def put(self, request, pk):
+        instance = self.get_object(pk)
+        serializer = InstanceDetailSerializer(instance, data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    @extend_schema(summary="删除实例",
+                   description="删除一个实例配置")
+    def delete(self, request, pk):
+        instance = self.get_object(pk)
+        instance.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+
+class TunnelList(generics.ListAPIView):
+    """
+    列出所有的tunnel或者创建一个新的tunnel配置
+    """
+    pagination_class = CustomizedPagination
+    serializer_class = TunnelSerializer
+    queryset = Tunnel.objects.all().order_by('id')
+
+    @extend_schema(summary="隧道清单",
+                   request=TunnelSerializer,
+                   responses={200: TunnelSerializer},
+                   description="列出所有隧道（过滤，分页）")
+    def get(self, request):
+        tunnels = self.filter_queryset(self.queryset)
+        page_tunnels = self.paginate_queryset(queryset=tunnels)
+        serializer_obj = self.get_serializer(page_tunnels, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建隧道",
+                   request=TunnelSerializer,
+                   responses={201: TunnelSerializer},
+                   description="创建一个隧道配置")
+    def post(self, request):
+        serializer = TunnelSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class AliyunRdsList(generics.ListAPIView):
+    """
+    列出所有的AliyunRDS或者创建一个新的AliyunRDS配置
+    """
+    pagination_class = CustomizedPagination
+    serializer_class = AliyunRdsSerializer
+    queryset = AliyunRdsConfig.objects.all().select_related('ak').order_by('id')
+
+    @extend_schema(summary="AliyunRDS清单",
+                   request=AliyunRdsSerializer,
+                   responses={200: AliyunRdsSerializer},
+                   description="列出所有AliyunRDS（过滤，分页）")
+    def get(self, request):
+        aliyunrds = self.filter_queryset(self.queryset)
+        page_rds = self.paginate_queryset(queryset=aliyunrds)
+        serializer_obj = self.get_serializer(page_rds, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建AliyunRDS",
+                   request=AliyunRdsSerializer,
+                   responses={201: AliyunRdsSerializer},
+                   description="创建一个AliyunRDS配置（包含一个CloudAccessKey）")
+    def post(self, request):
+        serializer = AliyunRdsSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class InstanceResource(views.APIView):
+    """
+    获取实例内的资源信息，database、schema、table、column
+    """
+
+    @extend_schema(summary="实例资源",
+                   request=InstanceResourceSerializer,
+                   responses={200: InstanceResourceListSerializer},
+                   description="获取实例内的资源信息")
+    def post(self, request):
+        # 参数验证
+        serializer = InstanceResourceSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        instance_id = request.data['instance_id']
+        resource_type = request.data['resource_type']
+        db_name = request.data['db_name'] if 'db_name' in request.data.keys() else ''
+        schema_name = request.data['schema_name'] if 'schema_name' in request.data.keys() else ''
+        tb_name = request.data['tb_name'] if 'tb_name' in request.data.keys() else ''
+        instance = Instance.objects.get(pk=instance_id)
+
+        try:
+            # escape
+            db_name = MySQLdb.escape_string(db_name).decode('utf-8')
+            schema_name = MySQLdb.escape_string(schema_name).decode('utf-8')
+            tb_name = MySQLdb.escape_string(tb_name).decode('utf-8')
+
+            query_engine = get_engine(instance=instance)
+            if resource_type == 'database':
+                resource = query_engine.get_all_databases()
+            elif resource_type == 'schema' and db_name:
+                resource = query_engine.get_all_schemas(db_name=db_name)
+            elif resource_type == 'table' and db_name:
+                resource = query_engine.get_all_tables(db_name=db_name, schema_name=schema_name)
+            elif resource_type == 'column' and db_name and tb_name:
+                resource = query_engine.get_all_columns_by_tb(db_name=db_name, tb_name=tb_name, schema_name=schema_name)
+            else:
+                raise serializers.ValidationError({'errors': '不支持的资源类型或者参数不完整！'})
+        except Exception as msg:
+            raise serializers.ValidationError({'errors': msg})
+        else:
+            if resource.error:
+                raise serializers.ValidationError({'errors': resource.error})
+            else:
+                resource = {'count': len(resource.rows),
+                            'result': resource.rows}
+                serializer_obj = InstanceResourceListSerializer(resource)
+                return Response(serializer_obj.data)
diff --git a/sql_api/api_user.py b/sql_api/api_user.py
new file mode 100644
index 0000000000..61f606cb73
--- /dev/null
+++ b/sql_api/api_user.py
@@ -0,0 +1,205 @@
+from rest_framework import views, generics, status
+from rest_framework.response import Response
+from drf_spectacular.utils import extend_schema
+from .serializers import UserSerializer, UserDetailSerializer, GroupSerializer, ResourceGroupSerializer
+from .pagination import CustomizedPagination
+from .filters import UserFilter
+from django.contrib.auth.models import Group
+from django.http import Http404
+from sql.models import Users, ResourceGroup
+
+
+class UserList(generics.ListAPIView):
+    """
+    列出所有的user或者创建一个新的user
+    """
+    filterset_class = UserFilter
+    pagination_class = CustomizedPagination
+    serializer_class = UserSerializer
+    queryset = Users.objects.all().order_by('id')
+
+    @extend_schema(summary="用户清单",
+                   request=UserSerializer,
+                   responses={200: UserSerializer},
+                   description="列出所有用户（过滤，分页）")
+    def get(self, request):
+        users = self.filter_queryset(self.queryset)
+        page_user = self.paginate_queryset(queryset=users)
+        serializer_obj = self.get_serializer(page_user, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建用户",
+                   request=UserSerializer,
+                   responses={201: UserSerializer},
+                   description="创建一个用户")
+    def post(self, request):
+        serializer = UserSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class UserDetail(views.APIView):
+    """
+    用户操作
+    """
+    serializer_class = UserDetailSerializer
+
+    def get_object(self, pk):
+        try:
+            return Users.objects.get(pk=pk)
+        except Users.DoesNotExist:
+            raise Http404
+
+    @extend_schema(summary="更新用户",
+                   request=UserDetailSerializer,
+                   responses={200: UserDetailSerializer},
+                   description="更新一个用户")
+    def put(self, request, pk):
+        user = self.get_object(pk)
+        serializer = UserDetailSerializer(user, data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    @extend_schema(summary="删除用户",
+                   description="删除一个用户")
+    def delete(self, request, pk):
+        user = self.get_object(pk)
+        user.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+
+class GroupList(generics.ListAPIView):
+    """
+    列出所有的group或者创建一个新的group
+    """
+    pagination_class = CustomizedPagination
+    serializer_class = GroupSerializer
+    queryset = Group.objects.all().order_by('id')
+
+    @extend_schema(summary="用户组清单",
+                   request=GroupSerializer,
+                   responses={200: GroupSerializer},
+                   description="列出所有用户组（过滤，分页）")
+    def get(self, request):
+        groups = self.filter_queryset(self.queryset)
+        page_groups = self.paginate_queryset(queryset=groups)
+        serializer_obj = self.get_serializer(page_groups, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建用户组",
+                   request=GroupSerializer,
+                   responses={201: GroupSerializer},
+                   description="创建一个用户组")
+    def post(self, request):
+        serializer = GroupSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class GroupDetail(views.APIView):
+    """
+    用户组操作
+    """
+    serializer_class = GroupSerializer
+
+    def get_object(self, pk):
+        try:
+            return Group.objects.get(pk=pk)
+        except Group.DoesNotExist:
+            raise Http404
+
+    @extend_schema(summary="更新用户组",
+                   request=GroupSerializer,
+                   responses={200: GroupSerializer},
+                   description="更新一个用户组")
+    def put(self, request, pk):
+        group = self.get_object(pk)
+        serializer = GroupSerializer(group, data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    @extend_schema(summary="删除用户组",
+                   description="删除一个用户组")
+    def delete(self, request, pk):
+        group = self.get_object(pk)
+        group.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
+
+class ResourceGroupList(generics.ListAPIView):
+    """
+    列出所有的resourcegroup或者创建一个新的resourcegroup
+    """
+    pagination_class = CustomizedPagination
+    serializer_class = ResourceGroupSerializer
+    queryset = ResourceGroup.objects.all().order_by('group_id')
+
+    @extend_schema(summary="资源组清单",
+                   request=ResourceGroupSerializer,
+                   responses={200: ResourceGroupSerializer},
+                   description="列出所有资源组（过滤，分页）")
+    def get(self, request):
+        groups = self.filter_queryset(self.queryset)
+        page_groups = self.paginate_queryset(queryset=groups)
+        serializer_obj = self.get_serializer(page_groups, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="创建资源组",
+                   request=ResourceGroupSerializer,
+                   responses={201: ResourceGroupSerializer},
+                   description="创建一个资源组")
+    def post(self, request):
+        serializer = ResourceGroupSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class ResourceGroupDetail(views.APIView):
+    """
+    资源组操作
+    """
+    serializer_class = ResourceGroupSerializer
+
+    def get_object(self, pk):
+        try:
+            return ResourceGroup.objects.get(pk=pk)
+        except ResourceGroup.DoesNotExist:
+            raise Http404
+
+    @extend_schema(summary="更新资源组",
+                   request=ResourceGroupSerializer,
+                   responses={200: ResourceGroupSerializer},
+                   description="更新一个资源组")
+    def put(self, request, pk):
+        group = self.get_object(pk)
+        serializer = ResourceGroupSerializer(group, data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+    @extend_schema(summary="删除资源组",
+                   description="删除一个资源组")
+    def delete(self, request, pk):
+        group = self.get_object(pk)
+        group.delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
diff --git a/sql_api/api_workflow.py b/sql_api/api_workflow.py
new file mode 100644
index 0000000000..3575fcc6e6
--- /dev/null
+++ b/sql_api/api_workflow.py
@@ -0,0 +1,440 @@
+from rest_framework import views, generics, status, serializers
+from rest_framework.response import Response
+from drf_spectacular.utils import extend_schema
+from .serializers import WorkflowContentSerializer, ExecuteCheckSerializer, \
+    ExecuteCheckResultSerializer, WorkflowAuditSerializer, WorkflowAuditListSerializer, \
+    WorkflowLogSerializer, WorkflowLogListSerializer, AuditWorkflowSerializer, ExecuteWorkflowSerializer
+from .pagination import CustomizedPagination
+from .filters import WorkflowFilter, WorkflowAuditFilter
+from sql.models import SqlWorkflow, SqlWorkflowContent, Instance, 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')
+
+
+class ExecuteCheck(views.APIView):
+    @extend_schema(summary="SQL检查",
+                   request=ExecuteCheckSerializer,
+                   responses={200: ExecuteCheckResultSerializer},
+                   description="对提供的SQL进行语法检查")
+    def post(self, request):
+        # 参数验证
+        serializer = ExecuteCheckSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        instance = Instance.objects.get(pk=request.data['instance_id'])
+        check_engine = get_engine(instance=instance)
+        check_result = check_engine.execute_check(db_name=request.data['db_name'],
+                                                  sql=request.data['full_sql'].strip())
+        review_result_list = []
+        for r in check_result.rows:
+            review_result_list += [r.__dict__]
+        check_result.rows = review_result_list
+        serializer_obj = ExecuteCheckResultSerializer(check_result)
+        return Response(serializer_obj.data)
+
+
+class WorkflowList(generics.ListAPIView):
+    """
+    列出所有的workflow或者提交一条新的workflow
+    """
+    filterset_class = WorkflowFilter
+    pagination_class = CustomizedPagination
+    serializer_class = WorkflowContentSerializer
+    queryset = SqlWorkflowContent.objects.all().select_related('workflow').order_by('-id')
+
+    @extend_schema(summary="SQL上线工单清单",
+                   request=WorkflowContentSerializer,
+                   responses={200: WorkflowContentSerializer},
+                   description="列出所有SQL上线工单（过滤，分页）")
+    def get(self, request):
+        workflows = self.filter_queryset(self.queryset)
+        page_wf = self.paginate_queryset(queryset=workflows)
+        serializer_obj = self.get_serializer(page_wf, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+    @extend_schema(summary="提交SQL上线工单",
+                   request=WorkflowContentSerializer,
+                   responses={201: WorkflowContentSerializer},
+                   description="提交一条SQL上线工单")
+    def post(self, request):
+        serializer = WorkflowContentSerializer(data=request.data)
+        if serializer.is_valid():
+            serializer.save()
+            return Response(serializer.data, status=status.HTTP_201_CREATED)
+        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+
+class WorkflowAuditList(generics.ListAPIView):
+    """
+    列出指定用户当前待自己审核的工单
+    """
+    filterset_class = WorkflowAuditFilter
+    pagination_class = CustomizedPagination
+    serializer_class = WorkflowAuditListSerializer
+    queryset = WorkflowAudit.objects.filter(
+        current_status=WorkflowDict.workflow_status['audit_wait']).order_by('-audit_id')
+
+    @extend_schema(exclude=True)
+    def get(self, request):
+        return Response({"detail": "方法 “GET” 不被允许。"})
+
+    @extend_schema(summary="待审核清单",
+                   request=WorkflowAuditSerializer,
+                   responses={200: WorkflowAuditListSerializer},
+                   description="列出指定用户待审核清单（过滤，分页）")
+    def post(self, request):
+        # 参数验证
+        serializer = WorkflowAuditSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        # 先获取用户所在资源组列表
+        user = Users.objects.get(username=request.data['engineer'])
+        group_list = user_groups(user)
+        group_ids = [group.group_id for group in group_list]
+
+        # 再获取用户所在权限组列表
+        if user.is_superuser:
+            auth_group_ids = [group.id for group in Group.objects.all()]
+        else:
+            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'],
+                                             group_id__in=group_ids,
+                                             current_audit__in=auth_group_ids)
+        audit = self.filter_queryset(self.queryset)
+        page_audit = self.paginate_queryset(queryset=audit)
+        serializer_obj = self.get_serializer(page_audit, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
+
+
+class AuditWorkflow(views.APIView):
+    """
+    审核workflow，包括查询权限申请、SQL上线申请、数据归档申请
+    """
+
+    @extend_schema(summary="审核工单",
+                   request=AuditWorkflowSerializer,
+                   description="审核一条工单（通过或终止）")
+    def post(self, request):
+        # 参数验证
+        serializer = AuditWorkflowSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        audit_type = request.data['audit_type']
+        workflow_type = request.data['workflow_type']
+        workflow_id = request.data['workflow_id']
+        audit_remark = request.data['audit_remark']
+        engineer = request.data['engineer']
+        user = Users.objects.get(username=engineer)
+
+        # 审核查询权限申请
+        if workflow_type == 1:
+            audit_status = 1 if audit_type == 'pass' else 2
+
+            if audit_remark is None:
+                audit_remark = ''
+
+            if Audit.can_review(user, workflow_id, workflow_type) is False:
+                raise serializers.ValidationError({"errors": "你无权操作当前工单！"})
+
+            # 使用事务保持数据一致性
+            try:
+                with transaction.atomic():
+                    audit_id = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                           workflow_type=WorkflowDict.workflow_type['query']).audit_id
+
+                    # 调用工作流接口审核
+                    audit_result = Audit.audit(audit_id, audit_status, user.username, audit_remark)
+
+                    # 按照审核结果更新业务表审核状态
+                    audit_detail = Audit.detail(audit_id)
+                    if audit_detail.workflow_type == WorkflowDict.workflow_type['query']:
+                        # 更新业务表审核状态,插入权限信息
+                        _query_apply_audit_call_back(audit_detail.workflow_id, audit_result['data']['workflow_status'])
+
+            except Exception as msg:
+                logger.error(traceback.format_exc())
+                raise serializers.ValidationError({'errors': msg})
+            else:
+                # 消息通知
+                async_task(notify_for_audit, audit_id=audit_id, audit_remark=audit_remark, timeout=60,
+                           task_name=f'query-priv-audit-{workflow_id}')
+                return Response({'msg': 'passed'}) if audit_type == 'pass' else Response({'msg': 'canceled'})
+        # 审核SQL上线申请
+        elif workflow_type == 2:
+            # SQL上线申请通过
+            if audit_type == 'pass':
+                # 权限验证
+                if Audit.can_review(user, workflow_id, workflow_type) is False:
+                    raise serializers.ValidationError({"errors": "你无权操作当前工单！"})
+
+                # 使用事务保持数据一致性
+                try:
+                    with transaction.atomic():
+                        # 调用工作流接口审核
+                        audit_id = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                               workflow_type=WorkflowDict.workflow_type[
+                                                                   'sqlreview']).audit_id
+                        audit_result = Audit.audit(audit_id, WorkflowDict.workflow_status['audit_success'],
+                                                   user.username, audit_remark)
+
+                        # 按照审核结果更新业务表审核状态
+                        if audit_result['data']['workflow_status'] == WorkflowDict.workflow_status['audit_success']:
+                            # 将流程状态修改为审核通过
+                            SqlWorkflow(id=workflow_id, status='workflow_review_pass').save(update_fields=['status'])
+                except Exception as msg:
+                    logger.error(traceback.format_exc())
+                    raise serializers.ValidationError({'errors': msg})
+                else:
+                    # 开启了Pass阶段通知参数才发送消息通知
+                    sys_config = SysConfig()
+                    is_notified = 'Pass' in sys_config.get('notify_phase_control').split(',') \
+                        if sys_config.get('notify_phase_control') else True
+                    if is_notified:
+                        async_task(notify_for_audit, audit_id=audit_id, audit_remark=audit_remark, timeout=60,
+                                   task_name=f'sqlreview-pass-{workflow_id}')
+                    return Response({'msg': 'passed'})
+            # SQL上线申请驳回/取消
+            elif audit_type == 'cancel':
+                workflow_detail = SqlWorkflow.objects.get(id=workflow_id)
+
+                if audit_remark is None:
+                    raise serializers.ValidationError({"errors": "终止原因不能为空"})
+
+                if can_cancel(user, workflow_id) is False:
+                    raise serializers.ValidationError({"errors": "你无权操作当前工单！"})
+
+                # 使用事务保持数据一致性
+                try:
+                    with transaction.atomic():
+                        # 调用工作流接口取消或者驳回
+                        audit_id = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                               workflow_type=WorkflowDict.workflow_type[
+                                                                   'sqlreview']).audit_id
+                        # 仅待审核的需要调用工作流，审核通过的不需要
+                        if workflow_detail.status != 'workflow_manreviewing':
+                            # 增加工单日志
+                            if user.username == workflow_detail.engineer:
+                                Audit.add_log(audit_id=audit_id,
+                                              operation_type=3,
+                                              operation_type_desc='取消执行',
+                                              operation_info="取消原因：{}".format(audit_remark),
+                                              operator=user.username,
+                                              operator_display=user.display
+                                              )
+                            else:
+                                Audit.add_log(audit_id=audit_id,
+                                              operation_type=2,
+                                              operation_type_desc='审批不通过',
+                                              operation_info="审批备注：{}".format(audit_remark),
+                                              operator=user.username,
+                                              operator_display=user.display
+                                              )
+                        else:
+                            if user.username == workflow_detail.engineer:
+                                Audit.audit(audit_id,
+                                            WorkflowDict.workflow_status['audit_abort'],
+                                            user.username, audit_remark)
+                            # 非提交人需要校验审核权限
+                            elif user.has_perm('sql.sql_review'):
+                                Audit.audit(audit_id,
+                                            WorkflowDict.workflow_status['audit_reject'],
+                                            user.username, audit_remark)
+                            else:
+                                raise serializers.ValidationError({"errors": "Permission Denied"})
+
+                        # 删除定时执行task
+                        if workflow_detail.status == 'workflow_timingtask':
+                            schedule_name = f"sqlreview-timing-{workflow_id}"
+                            del_schedule(schedule_name)
+                        # 将流程状态修改为人工终止流程
+                        workflow_detail.status = 'workflow_abort'
+                        workflow_detail.save()
+                except Exception as msg:
+                    logger.error(f"取消工单报错，错误信息：{traceback.format_exc()}")
+                    raise serializers.ValidationError({'errors': msg})
+                else:
+                    # 发送取消、驳回通知，开启了Cancel阶段通知参数才发送消息通知
+                    sys_config = SysConfig()
+                    is_notified = 'Cancel' in sys_config.get('notify_phase_control').split(',') \
+                        if sys_config.get('notify_phase_control') else True
+                    if is_notified:
+                        audit_detail = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                                   workflow_type=WorkflowDict.workflow_type[
+                                                                       'sqlreview'])
+                        if audit_detail.current_status in (
+                                WorkflowDict.workflow_status['audit_abort'],
+                                WorkflowDict.workflow_status['audit_reject']):
+                            async_task(notify_for_audit, audit_id=audit_detail.audit_id, audit_remark=audit_remark,
+                                       timeout=60,
+                                       task_name=f'sqlreview-cancel-{workflow_id}')
+                    return Response({'msg': 'canceled'})
+        # 审核数据归档申请
+        elif workflow_type == 3:
+            audit_status = 1 if audit_type == 'pass' else 2
+
+            if audit_remark is None:
+                audit_remark = ''
+
+            if Audit.can_review(user, workflow_id, workflow_type) is False:
+                raise serializers.ValidationError({"errors": "你无权操作当前工单！"})
+
+            # 使用事务保持数据一致性
+            try:
+                with transaction.atomic():
+                    audit_id = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                           workflow_type=WorkflowDict.workflow_type['archive']).audit_id
+
+                    # 调用工作流插入审核信息，更新业务表审核状态
+                    audit_status = Audit.audit(audit_id, audit_status, user.username, audit_remark)['data'][
+                        'workflow_status']
+                    ArchiveConfig(id=workflow_id,
+                                  status=audit_status,
+                                  state=True if audit_status == WorkflowDict.workflow_status['audit_success'] else False
+                                  ).save(update_fields=['status', 'state'])
+            except Exception as msg:
+                logger.error(traceback.format_exc())
+                raise serializers.ValidationError({'errors': msg})
+            else:
+                # 消息通知
+                async_task(notify_for_audit, audit_id=audit_id, audit_remark=audit_remark, timeout=60,
+                           task_name=f'archive-audit-{workflow_id}')
+                return Response({'msg': 'passed'}) if audit_type == 'pass' else Response({'msg': 'canceled'})
+
+
+class ExecuteWorkflow(views.APIView):
+    """
+    执行workflow，包括SQL上线工单、数据归档工单
+    """
+
+    @extend_schema(summary="执行工单",
+                   request=ExecuteWorkflowSerializer,
+                   description="执行一条工单")
+    def post(self, request):
+        # 参数验证
+        serializer = ExecuteWorkflowSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        workflow_type = request.data['workflow_type']
+        workflow_id = request.data['workflow_id']
+
+        # 执行SQL上线工单
+        if workflow_type == 2:
+            mode = request.data['mode']
+            engineer = request.data['engineer']
+            user = Users.objects.get(username=engineer)
+
+            # 校验多个权限
+            if not (user.has_perm('sql.sql_execute') or user.has_perm('sql.sql_execute_for_resource_group')):
+                raise serializers.ValidationError({"errors": "你无权执行当前工单！"})
+
+            if can_execute(user, workflow_id) is False:
+                raise serializers.ValidationError({"errors": "你无权执行当前工单！"})
+
+            if on_correct_time_period(workflow_id) is False:
+                raise serializers.ValidationError({"errors": "不在可执行时间范围内，如果需要修改执行时间请重新提交工单!"})
+
+            # 获取审核信息
+            audit_id = Audit.detail_by_workflow_id(workflow_id=workflow_id,
+                                                   workflow_type=WorkflowDict.workflow_type['sqlreview']).audit_id
+
+            # 交由系统执行
+            if mode == "auto":
+                # 修改工单状态为排队中
+                SqlWorkflow(id=workflow_id, status="workflow_queuing").save(update_fields=['status'])
+                # 删除定时执行任务
+                schedule_name = f"sqlreview-timing-{workflow_id}"
+                del_schedule(schedule_name)
+                # 加入执行队列
+                async_task('sql.utils.execute_sql.execute', workflow_id, user,
+                           hook='sql.utils.execute_sql.execute_callback',
+                           timeout=-1, task_name=f'sqlreview-execute-{workflow_id}')
+                # 增加工单日志
+                Audit.add_log(audit_id=audit_id,
+                              operation_type=5,
+                              operation_type_desc='执行工单',
+                              operation_info='工单执行排队中',
+                              operator=user.username,
+                              operator_display=user.display)
+
+            # 线下手工执行
+            elif mode == "manual":
+                # 将流程状态修改为执行结束
+                SqlWorkflow(id=workflow_id, status="workflow_finish", finish_time=datetime.datetime.now()
+                            ).save(update_fields=['status', 'finish_time'])
+                # 增加工单日志
+                Audit.add_log(audit_id=audit_id,
+                              operation_type=6,
+                              operation_type_desc='手工工单',
+                              operation_info='确认手工执行结束',
+                              operator=user.username,
+                              operator_display=user.display)
+                # 开启了Execute阶段通知参数才发送消息通知
+                sys_config = SysConfig()
+                is_notified = 'Execute' in sys_config.get('notify_phase_control').split(',') \
+                    if sys_config.get('notify_phase_control') else True
+                if is_notified:
+                    notify_for_execute(SqlWorkflow.objects.get(id=workflow_id))
+        # 执行数据归档工单
+        elif workflow_type == 3:
+            async_task('sql.archiver.archive', workflow_id, timeout=-1, task_name=f'archive-{workflow_id}')
+
+        return Response({'msg': '开始执行，执行结果请到工单详情页查看'})
+
+
+class WorkflowLogList(generics.ListAPIView):
+    """
+    获取某个工单的日志
+    """
+    pagination_class = CustomizedPagination
+    serializer_class = WorkflowLogListSerializer
+    queryset = WorkflowLog.objects.all()
+
+    @extend_schema(exclude=True)
+    def get(self, request):
+        return Response({"detail": "方法 “GET” 不被允许。"})
+
+    @extend_schema(summary="工单日志",
+                   request=WorkflowLogSerializer,
+                   responses={200: WorkflowLogListSerializer},
+                   description="获取某个工单的日志（分页）")
+    def post(self, request):
+        # 参数验证
+        serializer = WorkflowLogSerializer(data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        audit_id = WorkflowAudit.objects.get(workflow_id=request.data['workflow_id'],
+                                             workflow_type=request.data['workflow_type']).audit_id
+        workflow_logs = self.queryset.filter(audit_id=audit_id).order_by('-id')
+        page_log = self.paginate_queryset(queryset=workflow_logs)
+        serializer_obj = self.get_serializer(page_log, many=True)
+        data = {
+            'data': serializer_obj.data
+        }
+        return self.get_paginated_response(data)
diff --git a/sql_api/filters.py b/sql_api/filters.py
new file mode 100644
index 0000000000..db71b0f630
--- /dev/null
+++ b/sql_api/filters.py
@@ -0,0 +1,50 @@
+from django_filters import rest_framework as filters
+from sql.models import Users, Instance, SqlWorkflowContent, WorkflowAudit
+
+
+class UserFilter(filters.FilterSet):
+
+    class Meta:
+        model = Users
+        fields = {
+            'id': ['exact'],
+            'username': ['exact'],
+        }
+
+
+class InstanceFilter(filters.FilterSet):
+
+    class Meta:
+        model = Instance
+        fields = {
+            'id': ['exact'],
+            'instance_name': ['icontains'],
+            'db_type': ['exact'],
+            'host': ['exact'],
+        }
+
+
+class WorkflowFilter(filters.FilterSet):
+
+    class Meta:
+        model = SqlWorkflowContent
+        fields = {
+            'id': ['exact'],
+            'workflow_id': ['exact'],
+            'workflow__workflow_name': ['icontains'],
+            'workflow__instance_id': ['exact'],
+            'workflow__db_name': ['exact'],
+            'workflow__engineer': ['exact'],
+            'workflow__status': ['exact'],
+            'workflow__create_time': ['lt', 'gte'],
+        }
+
+
+class WorkflowAuditFilter(filters.FilterSet):
+
+    class Meta:
+        model = WorkflowAudit
+        fields = {
+            'workflow_title': ['icontains'],
+            'workflow_type': ['exact'],
+        }
diff --git a/sql_api/pagination.py b/sql_api/pagination.py
new file mode 100644
index 0000000000..510015965c
--- /dev/null
+++ b/sql_api/pagination.py
@@ -0,0 +1,22 @@
+from rest_framework.views import Response
+from rest_framework.pagination import PageNumberPagination
+from collections import OrderedDict
+from django.conf import settings
+
+
+class CustomizedPagination(PageNumberPagination):
+    """
+    自定义分页器
+    """
+    page_size = settings.REST_FRAMEWORK['PAGE_SIZE'] if 'PAGE_SIZE' in settings.REST_FRAMEWORK.keys() else 20
+    page_query_param = 'page'
+    page_size_query_param = 'size'
+    max_page_size = None
+
+    def get_paginated_response(self, data):
+        return Response(OrderedDict([
+            ('count', data.get('count', self.page.paginator.count)),
+            ('next', self.get_next_link()),
+            ('previous', self.get_previous_link()),
+            ('results', data.get('data', data))
+        ]))
diff --git a/sql_api/permissions.py b/sql_api/permissions.py
new file mode 100644
index 0000000000..f1d8f54d09
--- /dev/null
+++ b/sql_api/permissions.py
@@ -0,0 +1,15 @@
+from rest_framework import permissions
+from common.config import SysConfig
+
+
+class IsInUserWhitelist(permissions.BasePermission):
+    """
+    自定义权限，只允许白名单用户调用api
+    """
+    def has_permission(self, request, view):
+        config = SysConfig().get('api_user_whitelist')
+        user_list = config.split(',') if config else []
+        api_user_whitelist = [int(uid) for uid in user_list]
+
+        # 只有在api_user_whitelist参数中的用户才有权限
+        return request.user.id in api_user_whitelist
diff --git a/sql_api/serializers.py b/sql_api/serializers.py
new file mode 100644
index 0000000000..646cf28b93
--- /dev/null
+++ b/sql_api/serializers.py
@@ -0,0 +1,408 @@
+from rest_framework import serializers
+from sql.models import Users, Instance, Tunnel, AliyunRdsConfig, CloudAccessKey, \
+    SqlWorkflow, SqlWorkflowContent, ResourceGroup, WorkflowAudit, WorkflowLog
+from django.contrib.auth.models import Group
+from django.contrib.auth.password_validation import validate_password
+from django.core.exceptions import ValidationError
+from django.db import transaction
+from django_q.tasks import async_task
+from sql.engines import get_engine
+from sql.utils.workflow_audit import Audit
+from sql.utils.resource_group import user_instances
+from sql.notify import notify_for_audit
+from common.utils.const import WorkflowDict
+from common.config import SysConfig
+import traceback
+import logging
+
+logger = logging.getLogger('default')
+
+
+class UserSerializer(serializers.ModelSerializer):
+    def create(self, validated_data):
+        user = Users(**validated_data)
+        user.set_password(validated_data["password"])
+        user.save()
+        return user
+
+    def validate_password(self, password):
+        try:
+            validate_password(password)
+        except ValidationError as msg:
+            raise serializers.ValidationError(msg)
+        return password
+
+    class Meta:
+        model = Users
+        fields = "__all__"
+        extra_kwargs = {
+            'password': {
+                'write_only': True
+            },
+            'display': {
+                'required': True
+            }
+        }
+
+
+class UserDetailSerializer(serializers.ModelSerializer):
+    def update(self, instance, validated_data):
+        for attr, value in validated_data.items():
+            if attr == 'password':
+                instance.set_password(value)
+            else:
+                setattr(instance, attr, value)
+        instance.save()
+        return instance
+
+    def validate_password(self, password):
+        try:
+            validate_password(password)
+        except ValidationError as msg:
+            raise serializers.ValidationError(msg)
+        return password
+
+    class Meta:
+        model = Users
+        fields = "__all__"
+        extra_kwargs = {
+            'password': {
+                'write_only': True,
+                'required': False
+            },
+            'username': {
+                'required': False
+            }
+        }
+
+
+class GroupSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = Group
+        fields = "__all__"
+
+
+class ResourceGroupSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = ResourceGroup
+        fields = "__all__"
+
+
+class InstanceSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = Instance
+        fields = "__all__"
+        extra_kwargs = {
+            'password': {
+                'write_only': True
+            }
+        }
+
+
+class InstanceDetailSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = Instance
+        fields = "__all__"
+        extra_kwargs = {
+            'password': {
+                'write_only': True
+            },
+            'instance_name': {
+                'required': False
+            },
+            'type': {
+                'required': False
+            },
+            'db_type': {
+                'required': False
+            },
+            'host': {
+                'required': False
+            }
+        }
+
+
+class TunnelSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = Tunnel
+        fields = "__all__"
+        write_only_fields = ['password', 'pkey', 'pkey_password']
+
+
+class CloudAccessKeySerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = CloudAccessKey
+        fields = "__all__"
+
+
+class AliyunRdsSerializer(serializers.ModelSerializer):
+    ak = CloudAccessKeySerializer()
+
+    def create(self, validated_data):
+        """创建包含accesskey的aliyunrds实例"""
+        rds_data = validated_data.pop('ak')
+
+        try:
+            with transaction.atomic():
+                ak = CloudAccessKey.objects.create(**rds_data)
+                rds = AliyunRdsConfig.objects.create(ak=ak, **validated_data)
+        except Exception as e:
+            logger.error(f"创建AliyunRds报错，错误信息：{traceback.format_exc()}")
+            raise serializers.ValidationError({'errors': str(e)})
+        else:
+            return rds
+
+    class Meta:
+        model = AliyunRdsConfig
+        fields = ('id', 'rds_dbinstanceid', 'is_enable', 'instance', 'ak')
+
+
+class InstanceResourceSerializer(serializers.Serializer):
+    instance_id = serializers.IntegerField(label='实例id')
+    resource_type = serializers.ChoiceField(choices=['database', 'schema', 'table', 'column'], label='资源类型')
+    db_name = serializers.CharField(required=False, label='数据库名')
+    schema_name = serializers.CharField(required=False, label='schema名')
+    tb_name = serializers.CharField(required=False, label='表名')
+
+    def validate(self, attrs):
+        instance_id = attrs.get('instance_id')
+
+        try:
+            Instance.objects.get(id=instance_id)
+        except Instance.DoesNotExist:
+            raise serializers.ValidationError({"errors": "不存在该实例"})
+
+        return attrs
+
+
+class InstanceResourceListSerializer(serializers.Serializer):
+    count = serializers.IntegerField()
+    result = serializers.ListField()
+
+
+class ExecuteCheckSerializer(serializers.Serializer):
+    instance_id = serializers.IntegerField(label='实例id')
+    db_name = serializers.CharField(label='数据库名')
+    full_sql = serializers.CharField(label='SQL内容')
+
+    def validate_instance_id(self, instance_id):
+        try:
+            Instance.objects.get(pk=instance_id)
+        except Instance.DoesNotExist:
+            raise serializers.ValidationError({"errors": f"不存在该实例：{instance_id}"})
+        return instance_id
+
+
+class ExecuteCheckResultSerializer(serializers.Serializer):
+    is_execute = serializers.BooleanField(read_only=True, default=False)
+    checked = serializers.CharField(read_only=True)
+    warning = serializers.CharField(read_only=True)
+    error = serializers.CharField(read_only=True)
+    warning_count = serializers.IntegerField(read_only=True)
+    error_count = serializers.IntegerField(read_only=True)
+    is_critical = serializers.BooleanField(read_only=True, default=False)
+    syntax_type = serializers.IntegerField(read_only=True)
+    rows = serializers.JSONField(read_only=True)
+    column_list = serializers.JSONField(read_only=True)
+    status = serializers.CharField(read_only=True)
+    affected_rows = serializers.IntegerField(read_only=True)
+
+
+class WorkflowSerializer(serializers.ModelSerializer):
+    def validate(self, attrs):
+        engineer = attrs.get('engineer')
+        group_id = attrs.get('group_id')
+
+        try:
+            Users.objects.get(username=engineer)
+        except Users.DoesNotExist:
+            raise serializers.ValidationError(f"不存在该用户：{engineer}")
+
+        try:
+            ResourceGroup.objects.get(pk=group_id)
+        except ResourceGroup.DoesNotExist:
+            raise serializers.ValidationError(f"不存在该资源组：{group_id}")
+
+        return attrs
+
+    class Meta:
+        model = SqlWorkflow
+        fields = "__all__"
+        read_only_fields = ['status', 'is_backup', 'syntax_type', 'audit_auth_groups', 'engineer_display',
+                            'group_name', 'finish_time', 'is_manual']
+        extra_kwargs = {
+            'demand_url': {
+                'required': False
+            }
+        }
+
+
+class WorkflowContentSerializer(serializers.ModelSerializer):
+    workflow = WorkflowSerializer()
+
+    def create(self, validated_data):
+        """使用原工单submit流程创建工单"""
+        workflow_data = validated_data.pop('workflow')
+        instance = workflow_data['instance']
+        sql_content = validated_data['sql_content'].strip()
+        user = Users.objects.get(username=workflow_data['engineer'])
+        group = ResourceGroup.objects.get(pk=workflow_data['group_id'])
+        active_user = Users.objects.filter(is_active=1)
+
+        # 验证组权限（用户是否在该组、该组是否有指定实例）
+        try:
+            user_instances(user, tag_codes=['can_write']).get(id=instance.id)
+        except instance.DoesNotExist:
+            raise serializers.ValidationError({'errors': '你所在组未关联该实例！'})
+
+        # 再次交给engine进行检测，防止绕过
+        try:
+            check_engine = get_engine(instance=instance)
+            check_result = check_engine.execute_check(db_name=workflow_data['db_name'],
+                                                      sql=sql_content)
+        except Exception as e:
+            raise serializers.ValidationError({'errors': str(e)})
+
+        # 未开启备份选项，并且engine支持备份，强制设置备份
+        is_backup = workflow_data['is_backup'] if 'is_backup' in workflow_data.keys() else False
+        sys_config = SysConfig()
+        if not sys_config.get('enable_backup_switch') and check_engine.auto_backup:
+            is_backup = True
+
+        # 按照系统配置确定是自动驳回还是放行
+        auto_review_wrong = sys_config.get('auto_review_wrong', '')  # 1表示出现警告就驳回，2和空表示出现错误才驳回
+        workflow_status = 'workflow_manreviewing'
+        if check_result.warning_count > 0 and auto_review_wrong == '1':
+            workflow_status = 'workflow_autoreviewwrong'
+        elif check_result.error_count > 0 and auto_review_wrong in ('', '1', '2'):
+            workflow_status = 'workflow_autoreviewwrong'
+
+        workflow_data.update(status=workflow_status,
+                             is_backup=is_backup,
+                             is_manual=0,
+                             syntax_type=check_result.syntax_type,
+                             engineer_display=user.display,
+                             group_name=group.group_name,
+                             audit_auth_groups=Audit.settings(workflow_data['group_id'],
+                                                              WorkflowDict.workflow_type['sqlreview']))
+        try:
+            with transaction.atomic():
+                workflow = SqlWorkflow.objects.create(**workflow_data)
+                validated_data['review_content'] = check_result.json()
+                workflow_content = SqlWorkflowContent.objects.create(workflow=workflow, **validated_data)
+                # 自动审核通过了，才调用工作流
+                if workflow_status == 'workflow_manreviewing':
+                    # 调用工作流插入审核信息, SQL上线权限申请workflow_type=2
+                    Audit.add(WorkflowDict.workflow_type['sqlreview'], workflow.id)
+        except Exception as e:
+            logger.error(f"提交工单报错，错误信息：{traceback.format_exc()}")
+            raise serializers.ValidationError({'errors': str(e)})
+        else:
+            # 自动审核通过且开启了Apply阶段通知参数才发送消息通知
+            is_notified = 'Apply' in sys_config.get('notify_phase_control').split(',') \
+                if sys_config.get('notify_phase_control') else True
+            if workflow_status == 'workflow_manreviewing' and is_notified:
+                # 获取审核信息
+                audit_id = Audit.detail_by_workflow_id(workflow_id=workflow.id,
+                                                       workflow_type=WorkflowDict.workflow_type['sqlreview']).audit_id
+                async_task(notify_for_audit, audit_id=audit_id, cc_users=active_user, timeout=60,
+                           task_name=f'sqlreview-submit-{workflow.id}')
+            return workflow_content
+
+    class Meta:
+        model = SqlWorkflowContent
+        fields = ('id', 'workflow_id', 'workflow', 'sql_content', 'review_content', 'execute_result')
+        read_only_fields = ['review_content', 'execute_result']
+
+
+class AuditWorkflowSerializer(serializers.Serializer):
+    engineer = serializers.CharField(label='操作用户')
+    workflow_id = serializers.IntegerField(label='工单id')
+    audit_remark = serializers.CharField(label='审批备注')
+    workflow_type = serializers.ChoiceField(choices=[1, 2, 3], label='工单类型：1-查询权限申请，2-SQL上线申请，3-数据归档申请')
+    audit_type = serializers.ChoiceField(choices=['pass', 'cancel'], label='审核类型')
+
+    def validate(self, attrs):
+        engineer = attrs.get('engineer')
+        workflow_id = attrs.get('workflow_id')
+        workflow_type = attrs.get('workflow_type')
+
+        try:
+            Users.objects.get(username=engineer)
+        except Users.DoesNotExist:
+            raise serializers.ValidationError({"errors": f"不存在该用户：{engineer}"})
+
+        try:
+            WorkflowAudit.objects.get(workflow_id=workflow_id, workflow_type=workflow_type)
+        except WorkflowAudit.DoesNotExist:
+            raise serializers.ValidationError({"errors": "不存在该工单"})
+
+        return attrs
+
+
+class WorkflowAuditSerializer(serializers.Serializer):
+    engineer = serializers.CharField(label='操作用户')
+
+    def validate_engineer(self, engineer):
+        try:
+            Users.objects.get(username=engineer)
+        except Users.DoesNotExist:
+            raise serializers.ValidationError({"errors": f"不存在该用户：{engineer}"})
+        return engineer
+
+
+class WorkflowAuditListSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = WorkflowAudit
+        exclude = ['group_id', 'workflow_id', 'workflow_remark', 'next_audit', 'create_user', 'sys_time']
+
+
+class WorkflowLogSerializer(serializers.Serializer):
+    workflow_id = serializers.IntegerField(label='工单id')
+    workflow_type = serializers.ChoiceField(choices=[1, 2, 3], label='工单类型：1-查询权限申请，2-SQL上线申请，3-数据归档申请')
+
+
+class WorkflowLogListSerializer(serializers.ModelSerializer):
+
+    class Meta:
+        model = WorkflowLog
+        fields = ['operation_type_desc', 'operation_info', 'operator_display', 'operation_time']
+
+
+class ExecuteWorkflowSerializer(serializers.Serializer):
+    engineer = serializers.CharField(required=False, label='操作用户')
+    workflow_id = serializers.IntegerField(label='工单id')
+    workflow_type = serializers.ChoiceField(choices=[2, 3], label='工单类型：1-查询权限申请，2-SQL上线申请，3-数据归档申请')
+    mode = serializers.ChoiceField(choices=['auto', 'manual'], label='执行模式：auto-线上执行，manual-已手动执行', required=False)
+
+    def validate(self, attrs):
+        engineer = attrs.get('engineer')
+        workflow_id = attrs.get('workflow_id')
+        workflow_type = attrs.get('workflow_type')
+        mode = attrs.get('mode')
+
+        # SQL上线工单的mode和engineer为必需字段
+        if workflow_type == 2:
+            if not mode:
+                raise serializers.ValidationError({"errors": "缺少 mode"})
+            if not engineer:
+                raise serializers.ValidationError({"errors": "缺少 engineer"})
+
+            try:
+                Users.objects.get(username=engineer)
+            except Users.DoesNotExist:
+                raise serializers.ValidationError({"errors": f"不存在该用户：{engineer}"})
+
+        try:
+            WorkflowAudit.objects.get(workflow_id=workflow_id, workflow_type=workflow_type)
+        except WorkflowAudit.DoesNotExist:
+            raise serializers.ValidationError({"errors": "不存在该工单"})
+
+        return attrs
diff --git a/sql_api/tests.py b/sql_api/tests.py
index fa7e864884..079643f39e 100644
--- a/sql_api/tests.py
+++ b/sql_api/tests.py
@@ -1,5 +1,13 @@
+from datetime import datetime, timedelta
 from django.test import TestCase
 from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group, Permission
+from rest_framework.test import APITestCase
+from rest_framework import status
+from common.config import SysConfig
+from sql.models import ResourceGroup, Instance, AliyunRdsConfig, CloudAccessKey, Tunnel, \
+    SqlWorkflow, SqlWorkflowContent, WorkflowAudit, WorkflowLog, InstanceTag, WorkflowAuditSetting
+import json
 
 User = get_user_model()
 
@@ -21,3 +29,386 @@ def test_debug_api(self):
         r = self.client.get('/api/debug')
         r_json = r.json()
         self.assertIsInstance(r_json['archery']['version'], str)
+
+
+class TestUser(APITestCase):
+    """测试用户相关接口"""
+
+    def setUp(self):
+        self.user = User(username='test_user', display='测试用户', is_active=True)
+        self.user.set_password('test_password')
+        self.user.save()
+        self.group = Group.objects.create(id=1, name='DBA')
+        self.res_group = ResourceGroup.objects.create(group_id=1, group_name='test')
+        r = self.client.post('/api/auth/token/', {'username': 'test_user', 'password': 'test_password'}, format='json')
+        self.token = r.data['access']
+        self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + self.token)
+        SysConfig().set('api_user_whitelist', self.user.id)
+
+    def tearDown(self):
+        self.user.delete()
+        self.group.delete()
+        self.res_group.delete()
+        SysConfig().purge()
+
+    def test_user_not_in_whitelist(self):
+        """测试api用户白名单参数"""
+        SysConfig().set('api_user_whitelist', '')
+        r = self.client.get('/api/v1/user/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_403_FORBIDDEN)
+        self.assertDictEqual(r.json(), {'detail': '您没有执行该操作的权限。'})
+
+    def test_get_user_list(self):
+        """测试获取用户清单"""
+        r = self.client.get('/api/v1/user/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_user(self):
+        """测试创建用户"""
+        json_data = {
+            'username': 'test_user2',
+            'password': 'test_password2',
+            'display': '测试用户2'
+        }
+        r = self.client.post('/api/v1/user/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['username'], 'test_user2')
+
+    def test_update_user(self):
+        """测试更新用户"""
+        json_data = {
+            'display': '更新中文名'
+        }
+        r = self.client.put(f'/api/v1/user/{self.user.id}/', json_data, format='json')
+        user = User.objects.get(pk=self.user.id)
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(user.display, '更新中文名')
+
+    def test_delete_user(self):
+        """测试删除用户"""
+        json_data = {
+            'username': 'test_user2',
+            'password': 'test_password2',
+            'display': '测试用户2'
+        }
+        r1 = self.client.post('/api/v1/user/', json_data, format='json')
+        r2 = self.client.delete(f'/api/v1/user/{r1.json()["id"]}/', format='json')
+        self.assertEqual(r2.status_code, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(User.objects.filter(username='test_user2').count(), 0)
+
+    def test_get_user_group_list(self):
+        """测试获取用户组清单"""
+        r = self.client.get('/api/v1/user/group/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_user_group(self):
+        """测试创建用户组"""
+        json_data = {
+            'name': 'RD'
+        }
+        r = self.client.post('/api/v1/user/group/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['name'], 'RD')
+
+    def test_update_user_group(self):
+        """测试更新用户组"""
+        json_data = {
+            'name': '更新用户组名称'
+        }
+        r = self.client.put(f'/api/v1/user/group/{self.group.id}/', json_data, format='json')
+        group = Group.objects.get(pk=self.group.id)
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(group.name, '更新用户组名称')
+
+    def test_delete_user_group(self):
+        """测试删除用户组"""
+        r = self.client.delete(f'/api/v1/user/group/{self.group.id}/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(Group.objects.filter(name='DBA').count(), 0)
+
+    def test_get_resource_group_list(self):
+        """测试获取资源组清单"""
+        r = self.client.get('/api/v1/user/resourcegroup/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_resource_group(self):
+        """测试创建资源组"""
+        json_data = {
+            'group_name': 'prod',
+            'ding_webhook': 'https://oapi.dingtalk.com/robot/send?access_token=123'
+        }
+        r = self.client.post('/api/v1/user/resourcegroup/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['group_name'], 'prod')
+
+    def test_update_resource_group(self):
+        """测试更新资源组"""
+        json_data = {
+            'group_name': '更新资源组名称'
+        }
+        r = self.client.put(f'/api/v1/user/resourcegroup/{self.res_group.group_id}/', json_data, format='json')
+        group = ResourceGroup.objects.get(pk=self.res_group.group_id)
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(group.group_name, '更新资源组名称')
+
+    def test_delete_resource_group(self):
+        """测试删除资源组"""
+        r = self.client.delete(f'/api/v1/user/resourcegroup/{self.res_group.group_id}/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(Group.objects.filter(name='test').count(), 0)
+
+
+class TestInstance(APITestCase):
+    """测试实例相关接口"""
+
+    def setUp(self):
+        self.user = User(username='test_user', display='测试用户', is_active=True)
+        self.user.set_password('test_password')
+        self.user.save()
+        self.ins = Instance.objects.create(instance_name='some_ins', type='slave', db_type='mysql',
+                                           host='some_host', port=3306, user='ins_user', password='some_str')
+        self.ak = CloudAccessKey.objects.create(type='aliyun', key_id='abc', key_secret='abc')
+        self.rds = AliyunRdsConfig.objects.create(rds_dbinstanceid='abc', ak_id=self.ak.id, instance=self.ins)
+        self.tunnel = Tunnel.objects.create(tunnel_name='one_tunnel', host='one_host', port=22)
+        r = self.client.post('/api/auth/token/', {'username': 'test_user', 'password': 'test_password'}, format='json')
+        self.token = r.data['access']
+        self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + self.token)
+        SysConfig().set('api_user_whitelist', self.user.id)
+
+    def tearDown(self):
+        self.user.delete()
+        Instance.objects.all().delete()
+        AliyunRdsConfig.objects.all().delete()
+        CloudAccessKey.objects.all().delete()
+        Tunnel.objects.all().delete()
+        SysConfig().purge()
+
+    def test_get_instance_list(self):
+        """测试获取实例清单"""
+        r = self.client.get('/api/v1/instance/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_instance(self):
+        """测试创建实例"""
+        json_data = {
+            'instance_name': 'test_ins',
+            'type': 'master',
+            'db_type': 'mysql',
+            'host': 'some_host',
+            'port': 3306
+        }
+        r = self.client.post('/api/v1/instance/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['instance_name'], 'test_ins')
+
+    def test_update_instance(self):
+        """测试更新实例"""
+        json_data = {
+            'instance_name': '更新实例名称'
+        }
+        r = self.client.put(f'/api/v1/instance/{self.ins.id}/', json_data, format='json')
+        ins = Instance.objects.get(pk=self.ins.id)
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(ins.instance_name, '更新实例名称')
+
+    def test_delete_instance(self):
+        """测试删除实例"""
+        r = self.client.delete(f'/api/v1/instance/{self.ins.id}/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(Instance.objects.filter(instance_name='some_ins').count(), 0)
+
+    def test_get_aliyunrds_list(self):
+        """测试获取aliyunrds清单"""
+        r = self.client.get('/api/v1/instance/rds/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_aliyunrds(self):
+        """测试创建aliyunrds"""
+        ins = Instance.objects.create(instance_name='another_ins', type='slave', db_type='mysql',
+                                      host='another_host', port=3306)
+        json_data = {
+            "rds_dbinstanceid": "bbc",
+            "is_enable": True,
+            "instance": ins.id,
+            "ak": {
+                "type": "aliyun",
+                "key_id": "bbc",
+                "key_secret": "bbc",
+                "remark": "bbc"
+            }
+        }
+        r = self.client.post('/api/v1/instance/rds/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['rds_dbinstanceid'], 'bbc')
+
+    def test_get_tunnel_list(self):
+        """测试获取隧道清单"""
+        r = self.client.get('/api/v1/instance/tunnel/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_create_tunnel(self):
+        """测试创建隧道"""
+        json_data = {
+            "tunnel_name": "tunnel_test",
+            "host": "one_host",
+            "port": 22
+        }
+        r = self.client.post('/api/v1/instance/tunnel/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['tunnel_name'], 'tunnel_test')
+
+
+class TestWorkflow(APITestCase):
+    """测试工单相关接口"""
+
+    def setUp(self):
+        self.now = datetime.now()
+        self.group = Group.objects.create(id=1, name='DBA')
+        self.res_group = ResourceGroup.objects.create(group_id=1, group_name='test')
+        self.ins_tag = InstanceTag.objects.create(tag_code='can_write', active=1)
+        self.wfs = WorkflowAuditSetting.objects.create(group_id=self.res_group.group_id,
+                                                       workflow_type=2, audit_auth_groups=self.group.id)
+        can_execute_permission = Permission.objects.get(codename='sql_execute')
+        can_execute_resource_permission = Permission.objects.get(codename='sql_execute_for_resource_group')
+        can_review_permission = Permission.objects.get(codename='sql_review')
+        self.user = User(username='test_user', display='测试用户', is_active=True)
+        self.user.set_password('test_password')
+        self.user.save()
+        self.user.user_permissions.add(can_execute_permission, can_execute_resource_permission, can_review_permission)
+        self.user.groups.add(self.group.id)
+        self.user.resource_group.add(self.res_group.group_id)
+        self.ins = Instance.objects.create(instance_name='some_ins', type='slave', db_type='redis',
+                                           host='some_host', port=6379, user='ins_user', password='some_str')
+        self.ins.resource_group.add(self.res_group.group_id)
+        self.ins.instance_tag.add(self.ins_tag.id)
+        self.wf1 = SqlWorkflow.objects.create(
+            workflow_name='some_name',
+            group_id=1,
+            group_name='g1',
+            engineer=self.user.username,
+            engineer_display=self.user.display,
+            audit_auth_groups='1',
+            create_time=self.now - timedelta(days=1),
+            status='workflow_manreviewing',
+            is_backup=False,
+            instance=self.ins,
+            db_name='some_db',
+            syntax_type=1,
+        )
+        self.wfc1 = SqlWorkflowContent.objects.create(
+            workflow=self.wf1,
+            sql_content='some_sql',
+            execute_result=json.dumps([{
+                'id': 1,
+                'sql': 'some_content'
+            }])
+        )
+        self.audit1 = WorkflowAudit.objects.create(
+            group_id=1,
+            group_name='some_group',
+            workflow_id=self.wf1.id,
+            workflow_type=2,
+            workflow_title='申请标题',
+            workflow_remark='申请备注',
+            audit_auth_groups='1',
+            current_audit='1',
+            next_audit='-1',
+            current_status=0)
+        self.wl = WorkflowLog.objects.create(audit_id=self.audit1.audit_id,
+                                             operation_type=1)
+        r = self.client.post('/api/auth/token/', {'username': 'test_user', 'password': 'test_password'}, format='json')
+        self.token = r.data['access']
+        self.client.credentials(HTTP_AUTHORIZATION='Bearer ' + self.token)
+        SysConfig().set('api_user_whitelist', self.user.id)
+
+    def tearDown(self):
+        self.user.delete()
+        self.group.delete()
+        self.res_group.delete()
+        SqlWorkflowContent.objects.all().delete()
+        SqlWorkflow.objects.all().delete()
+        WorkflowAudit.objects.all().delete()
+        WorkflowLog.objects.all().delete()
+
+    def test_get_sql_workflow_list(self):
+        """测试获取SQL上线工单列表"""
+        r = self.client.get('/api/v1/workflow/', format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_get_audit_list(self):
+        """测试获取待审核工单列表"""
+        json_data = {
+            "engineer": "test_user"
+        }
+        r = self.client.post('/api/v1/workflow/auditlist/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_get_workflow_log_list(self):
+        """测试获工单日志"""
+        json_data = {
+            "workflow_id": self.wf1.id,
+            "workflow_type": self.audit1.workflow_type
+        }
+        r = self.client.post('/api/v1/workflow/log/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json()['count'], 1)
+
+    def test_submit_workflow(self):
+        """测试提交SQL上线工单"""
+        json_data = {
+            "workflow": {
+              "workflow_name": "上线工单1",
+              "demand_url": "test",
+              "group_id": 1,
+              "db_name": "test_db",
+              "engineer": self.user.username,
+              "instance": self.ins.id
+            },
+            "sql_content": "alter table abc add column note varchar(64);"
+        }
+        r = self.client.post('/api/v1/workflow/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(r.json()['workflow']['workflow_name'], '上线工单1')
+
+    def test_audit_workflow(self):
+        """测试审核工单"""
+        json_data = {
+            "engineer": self.user.username,
+            "workflow_id": self.wf1.id,
+            "audit_remark": "取消",
+            "workflow_type": self.audit1.workflow_type,
+            "audit_type": "cancel"
+        }
+        r = self.client.post('/api/v1/workflow/audit/', json_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json(), {'msg': 'canceled'})
+
+    def test_execute_workflow(self):
+        """测试执行工单"""
+        # 先审核
+        audit_data = {
+            "engineer": self.user.username,
+            "workflow_id": self.wf1.id,
+            "audit_remark": "通过",
+            "workflow_type": self.audit1.workflow_type,
+            "audit_type": "pass"
+        }
+        self.client.post('/api/v1/workflow/audit/', audit_data, format='json')
+        # 再执行
+        execute_data = {
+            "engineer": self.user.username,
+            "workflow_id": self.wf1.id,
+            "workflow_type": self.audit1.workflow_type,
+            "mode": "manual"
+        }
+        r = self.client.post('/api/v1/workflow/execute/', execute_data, format='json')
+        self.assertEqual(r.status_code, status.HTTP_200_OK)
+        self.assertEqual(r.json(), {'msg': '开始执行，执行结果请到工单详情页查看'})
diff --git a/sql_api/urls.py b/sql_api/urls.py
index f0b8e86610..d64089b613 100644
--- a/sql_api/urls.py
+++ b/sql_api/urls.py
@@ -1,7 +1,38 @@
-from django.urls import path
+from django.urls import path, include
+from django.conf.urls import url
 from sql_api import views
+from rest_framework import routers
+from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView, TokenVerifyView
+from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView, SpectacularSwaggerView
+from . import api_user, api_instance, api_workflow
+
+router = routers.DefaultRouter()
 
 urlpatterns = [
+    path('v1/', include(router.urls)),
+    path('auth/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'),
+    path('auth/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'),
+    path('auth/token/verify/', TokenVerifyView.as_view(), name='token_verify'),
+    path('schema/', SpectacularAPIView.as_view(), name='schema'),
+    path('swagger/', SpectacularSwaggerView.as_view(url_name='sql_api:schema'), name='swagger'),
+    path('redoc/', SpectacularRedocView.as_view(url_name='sql_api:schema'), name='redoc'),
+    path('v1/user/', api_user.UserList.as_view()),
+    url(r'^v1/user/(?P<pk>[0-9]+)/$', api_user.UserDetail.as_view()),
+    path('v1/user/group/', api_user.GroupList.as_view()),
+    url(r'^v1/user/group/(?P<pk>[0-9]+)/$', api_user.GroupDetail.as_view()),
+    path('v1/user/resourcegroup/', api_user.ResourceGroupList.as_view()),
+    url(r'^v1/user/resourcegroup/(?P<pk>[0-9]+)/$', api_user.ResourceGroupDetail.as_view()),
+    path('v1/instance/', api_instance.InstanceList.as_view()),
+    url(r'^v1/instance/(?P<pk>[0-9]+)/$', api_instance.InstanceDetail.as_view()),
+    path('v1/instance/resource/', api_instance.InstanceResource.as_view()),
+    path('v1/instance/tunnel/', api_instance.TunnelList.as_view()),
+    path('v1/instance/rds/', api_instance.AliyunRdsList.as_view()),
+    path('v1/workflow/', api_workflow.WorkflowList.as_view()),
+    path('v1/workflow/sqlcheck/', api_workflow.ExecuteCheck.as_view()),
+    path('v1/workflow/audit/', api_workflow.AuditWorkflow.as_view()),
+    path('v1/workflow/auditlist/', api_workflow.WorkflowAuditList.as_view()),
+    path('v1/workflow/execute/', api_workflow.ExecuteWorkflow.as_view()),
+    path('v1/workflow/log/', api_workflow.WorkflowLogList.as_view()),
     path('info', views.info),
     path('debug', views.debug),
     path('do_once/mirage', views.mirage)
diff --git a/src/init_sql/v1.8.4.sql b/src/init_sql/v1.8.4.sql
new file mode 100644
index 0000000000..b698710e78
--- /dev/null
+++ b/src/init_sql/v1.8.4.sql
@@ -0,0 +1,3 @@
+-- 新增openapi菜单权限
+set @content_type_id=(select id from django_content_type where app_label='sql' and model='permission');
+INSERT INTO auth_permission (name, content_type_id, codename) VALUES ('菜单 OpenAPI', @content_type_id, 'menu_openapi');