如何使用Django/Celery接收调用多个任务并收集任务状态

Question

我正在tasks.py中设置多个任务，并从views.py调用这些任务。我想在for循环中调用所有不同的任务，这样我就可以轻松地收集状态并构建进度条。我现在可以逐行调用所有任务(如下所示)。

以下是我的问题：如何在for循环中调用views.py中的不同任务？它总是给我一个错误"unicode没有属性delay()"。或者有更好的方法来收集不同任务的状态并根据它们构建进度条吗？

我尝试调用views.py中的函数，如下所示：对于范围(1，6)中的i：functionName="calculation"+str(i)functionName.delay(accountNumber)但这会产生如上所述的错误"unicode没有属性delay()">

我的猜测是，任务是从tasks.py导入到views.py

我当前的任务.py:

@shared_task
def calculation1(arg):
some action here
@shared_task
def calculation2(arg):
some action here
@shared_task
def calculation3(arg):
some action here
@shared_task
def calculation4(arg):
some action here
@shared_task
def calculation5(arg):
some action here

我的观点.py:

result_calculation1= calculation1.delay(accountNumber)
result_calculation2 = calculation2.delay(accountNumber)
result_calculation3 = calculation3.delay(accountNumber)
result_calculation4= calculation4.delay(accountNumber)
result_calculation5 = calculation5.delay(accountNumber)

我想在for循环中收集所有任务状态，这样我就可以建立进度条，但如果有其他关于收集任务状态和建立进度条的更好建议，那就太好了。

非常感谢您提前提供的帮助。

Answer 1

一旦构建了名称，就需要使用getattr()从tasks.py模块中检索函数：

from myapp import tasks  # Make sure you import the tasks module
for i in range (1, 6): 
functionName = "calculation" + str(i)  
task = getattr(tasks, functionName)  # Get the task by name from the tasks module

检索到任务函数后，您可以建立一个签名列表：

signatures = []
signatures.append(task.s(accountNumber))  # Add task signature

根据签名，您可以创建一个组并作为一个整体执行该组：

from celery import group
task_group = group(signatures)
group_result = group()  # Execute the group

从group_result中，您可以访问每个单独的任务结果，并围绕它构建进度条(也许可以迭代group_result中的结果，并检查每个结果的status)：

for result in group_result:
status = result.status
# Your progress bar logic...

综合起来：

from celery import group
from myapp import tasks  # Make sure you import the tasks module
signatures = []
for i in range (1, 6): 
functionName = "calculation" + str(i) 
task = getattr(tasks, functionName)  # Get the task from the tasks module
signatures.append(task.s(accountNumber))  # Add each task signature
task_group = group(signatures)
group_result = group()  # Execute the group
for result in group_result:
status = result.status
# Your progress bar logic...

Answer 2

您可以将任务放入一个组中，并可以接收GroupResult。你可以参考芹菜文档