feat: 任务派发测试

apps/develop/serializers.py

@@ -12,8 +12,10 @@ class GenerateVoiceSerializer(serializers.Serializer):
 
 
 class TestTaskSerializer(serializers.Serializer):
+    func = serializers.CharField(label='函数名称')
     args = serializers.ListField(child=serializers.CharField(), label='列表参数', required=False, allow_null=True)
     kwargs = serializers.JSONField(label="字典参数", required=False, allow_null=True)
+    delay = serializers.BooleanField(label="是否延迟执行", default=False)
 
 
 class CleanDataSerializer(serializers.Serializer):

apps/develop/views.py

@@ -22,6 +22,7 @@ from apps.third.models import TDevice
 from apps.utils.sms import send_sms
 from apps.utils.speech import generate_voice
 from apps.utils.tools import get_info_from_id
+import importlib
 
 from apps.utils.viewsets import CustomGenericViewSet
 from apps.utils.wx import wxClient
@@ -256,15 +257,15 @@ class TestViewSet(CustomGenericViewSet):
         serializer = TestTaskSerializer(data=request.data)
         serializer.is_valid(raise_exception=True)
         vdata = serializer.validated_data
-        ret = {}
-        # task = show.delay(*vdata.get('args', []), **vdata.get('kwargs', {}))
-        ret['task_id'] = task.task_id
-        # from celery.app.control import Control
-        # from server.celery import app
-        # celery_control = Control(app=app)
-        # # 关闭作业监控任务
-        # res = celery_control.revoke(task.task_id, terminate=True)
-        return Response(ret)
+        module, func = vdata['func'].rsplit(".", 1)
+        m = importlib.import_module(module)
+        f = getattr(m, func)
+        if vdata['delay']:
+            task = f.delay(*vdata.get('args', []), **vdata.get('kwargs', {}))  # 同步执行
+            return Response({'task_id': task.id})
+        else:
+            f(*vdata.get('args', []), **vdata.get('kwargs', {}))
+            return Response()
 
     @action(methods=['post'], detail=False, serializer_class=Serializer)
     def remind_event(self, request, pk=None):