sql工单orc操作迁移至api

Author: hhyo

sql/sql_workflow.py

@@ -137,24 +137,3 @@ def cancel(request):
             task_name=f"sqlreview-cancel-{workflow_id}",
         )
     return HttpResponseRedirect(reverse("sql:detail", args=(workflow_id,)))
-
-
-def osc_control(request):
-    """用于mysql控制osc执行"""
-    workflow_id = request.POST.get("workflow_id")
-    sqlsha1 = request.POST.get("sqlsha1")
-    command = request.POST.get("command")
-    workflow = SqlWorkflow.objects.get(id=workflow_id)
-    execute_engine = get_engine(workflow.instance)
-    try:
-        execute_result = execute_engine.osc_control(command=command, sqlsha1=sqlsha1)
-        rows = execute_result.to_dict()
-        error = execute_result.error
-    except Exception as e:
-        rows = []
-        error = str(e)
-    result = {"total": len(rows), "rows": rows, "msg": error}
-    return HttpResponse(
-        json.dumps(result, cls=ExtendJSONEncoder, bigint_as_string=True),
-        content_type="application/json",
-    )

sql/templates/detail.html

@@ -764,8 +764,9 @@ <h4 class="modal-title text-danger">修改可执行时间</h4>
             $('#osc_percent_list').bootstrapTable('destroy').bootstrapTable({
                 escape: true,
                 method: 'post',
-                contentType: "application/x-www-form-urlencoded",
-                url: "/inception/osc_control/",
+                contentType: "application/json",
+                url: "/api/v2/workflow/sql/" + workflow_id + "/mysql_osc_control/",
+                dataType: "json",
                 striped: true,                      //是否显示行间隔色
                 cache: false,                       //是否使用缓存，默认为true，所以一般情况下需要设置一下这个属性（*）
                 pagination: false,                   //是否显示分页（*）
@@ -808,11 +809,11 @@ <h4 class="modal-title text-danger">修改可执行时间</h4>
                 //请求服务数据时所传参数
                 queryParams:
                     function (params) {
-                        return {
+                        return JSON.stringify({
                             sqlsha1: $(obj).attr("sqlsha1"),
                             workflow_id: "{{ workflow_detail.id }}",
                             command: "get"
-                        }
+                        })
                     },
                 columns: [{
                     title: '',
@@ -887,11 +888,11 @@ <h4 class="modal-title text-danger">修改可执行时间</h4>
                     //发送请求
                     $.ajax({
                         type: "post",
-                        url: "/inception/osc_control/",
+                        contentType: "application/json",
+                        url: "/api/v2/workflow/sql/" + workflow_id + "/mysql_osc_control/",
                         dataType: "json",
                         data: {
                             sqlsha1: sqlsha1,
-                            workflow_id: "{{ workflow_detail.id }}",
                             command: command
                         },
                         complete: function () {

sql/tests.py

@@ -898,38 +898,6 @@ def testWorkflowCancelView(
         self.wf2.refresh_from_db()
         self.assertEqual("workflow_abort", self.wf2.status)
 
-    @patch("sql.sql_workflow.get_engine")
-    def test_osc_control(self, _get_engine):
-        """测试MySQL工单osc控制"""
-        c = Client()
-        c.force_login(self.superuser1)
-        request_data = {
-            "workflow_id": self.wf1.id,
-            "sqlsha1": "sqlsha1",
-            "command": "get",
-        }
-        _get_engine.return_value.osc_control.return_value = ResultSet()
-        r = c.post("/inception/osc_control/", data=request_data, follow=False)
-        self.assertDictEqual(
-            json.loads(r.content), {"total": 0, "rows": [], "msg": None}
-        )
-
-    @patch("sql.sql_workflow.get_engine")
-    def test_osc_control_exception(self, _get_engine):
-        """测试MySQL工单OSC控制异常"""
-        c = Client()
-        c.force_login(self.superuser1)
-        request_data = {
-            "workflow_id": self.wf1.id,
-            "sqlsha1": "sqlsha1",
-            "command": "get",
-        }
-        _get_engine.return_value.osc_control.side_effect = RuntimeError("RuntimeError")
-        r = c.post("/inception/osc_control/", data=request_data, follow=False)
-        self.assertDictEqual(
-            json.loads(r.content), {"total": 0, "rows": [], "msg": "RuntimeError"}
-        )
-
 
 class TestOptimize(TestCase):
     """

sql/urls.py

@@ -76,7 +76,6 @@
     path("audit_sqlworkflow/", views.audit_sqlworkflow),
     path("authenticate/", auth.authenticate_entry),
     path("del_sqlcronjob/", tasks.del_schedule),
-    path("inception/osc_control/", sql_workflow.osc_control),
     path("sql_analyze/generate/", sql_analyze.generate),
     path("sql_analyze/analyze/", sql_analyze.analyze),
     path("workflow/list/", workflow.lists),

sql_api/api_views/sql_workflow.py

@@ -10,7 +10,6 @@
 import django_filters
 from django.contrib.auth.decorators import permission_required
 from django.utils.decorators import method_decorator
-from drf_spectacular.types import OpenApiTypes
 from drf_spectacular.utils import extend_schema, extend_schema_view
 from rest_framework import serializers, viewsets, filters
 from rest_framework.decorators import action
@@ -30,6 +29,7 @@
     SqlWorkflowDetailSerializer,
     SqlWorkflowExecuteSerializer,
     SqlWorkflowTimingTaskSerializer,
+    SqlWorkflowMySQLOscControlSerializer,
 )
 
 
@@ -169,7 +169,7 @@ def alter_run_date(self, request, *args, **kwargs):
     @extend_schema(
         summary="立即执行工单",
         request=SqlWorkflowExecuteSerializer,
-        responses={200: OpenApiTypes.NUMBER},
+        responses={200: SqlWorkflowSerializer},
         description="通过工单ID执行工单",
     )
     @action(methods=["post"], detail=True)
@@ -179,12 +179,12 @@ def execute(self, request, *args, **kwargs):
         mode = request.data.get("mode")
         serializer = self.get_serializer()
         serializer.execute(obj, mode=mode, username=request.user.username)
-        return Response(obj.id)
+        return Response(serializer.data)
 
     @extend_schema(
         summary="设置定时执行工单",
         request=SqlWorkflowTimingTaskSerializer,
-        responses={200: OpenApiTypes.DATETIME},
+        responses={200: SqlWorkflowSerializer},
         description="通过工单ID执行工单",
     )
     @action(methods=["post"], detail=True)
@@ -196,4 +196,24 @@ def timing_task(self, request, *args, **kwargs):
         run_date = request.data.get("run_date")
         serializer = self.get_serializer()
         serializer.timing_task(obj, run_date=run_date, username=request.user.username)
-        return Response(run_date)
+        return Response(serializer.data)
+
+    @extend_schema(
+        summary="用于MySQL的大表DDL控制",
+        request=SqlWorkflowMySQLOscControlSerializer,
+        responses={200: SqlWorkflowSerializer},
+        description="控制pt-osc、gh-ost任务，用于执行进度获取，暂停、恢复、终止执行等",
+    )
+    @action(methods=["post"], detail=True)
+    def mysql_osc_control(self, request, *args, **kwargs):
+        obj = self.get_object()
+        SqlWorkflowMySQLOscControlSerializer(data=request.data).is_valid(
+            raise_exception=True
+        )
+        sqlsha1 = request.data.get("sqlsha1")
+        command = request.data.get("command")
+        serializer = self.get_serializer()
+        serializer.mysql_osc_control(
+            obj, sqlsha1=sqlsha1, command=command, username=request.user.username
+        )
+        return Response(serializer.data)

sql_api/serializers/sql_workflow.py

@@ -251,6 +251,19 @@ def timing_task(obj, run_date, username):
                 operator_display=user.display,
             )
 
+    @staticmethod
+    def mysql_osc_control(obj, sqlsha1, command):
+        """用于MySQL的大表DDL控制"""
+        execute_engine = get_engine(instance=obj.instance)
+        try:
+            execute_result = execute_engine.osc_control(command=command, sqlsha1=sqlsha1)
+            rows = execute_result.to_dict()
+            error = execute_result.error
+        except Exception as e:
+            rows = []
+            error = str(e)
+        return {"total": len(rows), "rows": rows, "msg": error}
+
     class Meta:
         model = SqlWorkflow
         fields = "__all__"
@@ -274,3 +287,14 @@ class SqlWorkflowTimingTaskSerializer(serializers.Serializer):
     run_date = serializers.DateTimeField(
         label="定时执行时间",
     )
+
+
+class SqlWorkflowMySQLOscControlSerializer(serializers.Serializer):
+    """用于MySQL的大表DDL控制"""
+
+    sqlsha1 = serializers.CharField(
+        label="SQL审核返回的sqlsha1信息",
+    )
+    command = serializers.ChoiceField(
+        choices=["get", "pause", "resume", "kill"], label="命令：get-获取详情，pause-暂停，resume-恢复，kill-终止"
+    )