feat: base 增加ctask_run将普通函数转celery执行

apps/utils/tasks.py

@@ -4,6 +4,7 @@ from celery import shared_task
 import logging
 from django.conf import settings
 from server.settings import get_sysconfig
+import importlib
 
 # 实例化myLogger
 myLogger = logging.getLogger('log')
@@ -20,7 +21,6 @@ def send_mail_task(**args):
         'recipient_list', [settings.EMAIL_HOST_USER])
     send_mail(**args)
 
-
 class CustomTask(Task):
     """
     自定义的任务回调
@@ -31,3 +31,12 @@ class CustomTask(Task):
         myLogger.error(detail)
         send_mail_task.delay(subject='task_error', message=detail)
         return super().on_failure(exc, task_id, args, kwargs, einfo)
+
+@shared_task(base=CustomTask)
+def ctask_run(func_str: str, *args, **kwargs):
+    """通用celery函数/将普通函数转为celery执行/也可直接运行
+    """
+    module, func = func_str.rsplit(".", 1)
+    m = importlib.import_module(module)
+    f = getattr(m, func)
+    f(*args, **kwargs)