python - 将Celery Group放入Flask应用程序中

我正在使用烧瓶，设置了一个简单的应用程序，并试图学习如何有效利用芹菜来工作，并在查询时显示结果。

在最基本的示例中，我通过task.delay(args)创建单个任务。然后，该对象让我提取该作业的ID，稍后可以通过单击其他端点进行查询。简单。

我的目标是模仿这一点，尽管要利用群体。阅读文档后，我发现组原语是惰性的，因此必须先实际调用它才能保存它。

我确定我的问题来自缺乏了解，但基本上是：
如果我的目标是能够通过Flask Pipeline异步并并行运行后台任务组，那么在以下约束条件下，我如何为该组检索.join()结果


请求Endpoint1（可能返回一个ID）
请求Endpotin2，传递ID以返回工作完成的结果


这是正确的方法吗？还是我应该思考的心态有所不同？

伪代码：

# From my apps init
celery_instance = Celery("module.modulename", backend = app.config['CELERY_RESULT_BACKEND'], broker = app.config['CELERY_BROKER_URL'])
celery_instance.conf.update(app.config)

<snip>

from celery import group
from app import celery_instance

@app.route("/status/domain/<id>", methods=['GET'])
def query(id):
  # Works for single job, not job group
  result = celery_instance.AsyncResult(id)
  ...

@app.route("/query/domain/<domain>", methods=['GET'])
def query_by_domain(domain):
  ...
  job = group([task1.delay(domain), task2.delay(domain)])
  return redirect(url_for('app.query', id=job.id), code=302)

最佳答案

您需要通过GroupResult.save()方法将组结果保存为：

@app.route("/query/domain/<domain>", methods=['GET'])
def query_by_domain(domain):
    ...
    job = group([task1.delay(domain), task2.delay(domain)])
    <b>job.save()</b>
    return redirect(url_for('app.query', id=job.id), code=302)

然后，您可以通过GroupResult方法获取GroupResult.restore()。

@app.route("/status/domain/<id>", methods=['GET'])
def query(id):
    # Works for single job, not job group
    <b>result = celery_instance.GroupResult.restore(group_id)</b>
    ...