在 Flask 中运行后台 Celery 任务答案

【问题标题】：Running background Celery task in Flask在 Flask 中运行后台 Celery 任务
【发布时间】：2013-08-08 17:52:04
【问题描述】：

问题已更新，包括已取得的进展

我有以下代码，并且我的 celery 任务启动良好，我只是不知道应该将异步结果存储在哪里，以便稍后再查看

#!/usr/bin/env python

"""Page views."""

from flask import render_template, request
from flask import Flask

from celerytest import add

from time import sleep


app = Flask(__name__)

async_res = []

@app.route('/', methods=['GET', 'POST'])
def run():
    if request.method == 'GET':
        return render_template("template.html")
    else:
        form = request.form
        n1 = str(form.get("n1"))
        n2 = str(form.get("n2"))
        aysnc_res = add.delay(n1,n2)
        return render_template("loading.html")

@app.route('/loading')
def check_if_complete():
    if async_res.ready() == True:
        return render_template("template2.html", val=async_res.get())
    else:
        sleep(5)
        return render_template("loading.html")


if __name__ == '__main__':
    app.run()

似乎在我的烧瓶应用程序中将async_res 存储为全局变量会导致服务器错误。那么存储结果的最佳方法是什么，以便我可以在“加载页面”中检查它

【问题讨论】：

通常情况下，您不会从 Flask 内部启动 Celery worker。您将在启动 Flask 时单独启动 Celery worker（参见 Worker Guide）。当任务运行时，您将启动一个如Task Guide 所示的任务。
@MarkHildreth 仅部分回答了我如何启动 Celery 任务的问题，我如何知道它何时完成并强制页面呈现结果？
您启动了我链接到的任务指南中所示的任务。您知道它已完成，因为您的任务在最后一步进行了一些更改（例如更改数据库中记录的状态标志）。您可以在 Web 请求中检查该状态标志。如何强制页面重新加载完全取决于您；有很多不同的方法（其中一些我在previous answer 中列出）。我建议您单独玩 Celery，直到您了解它可以做什么。
@MarkHildreth 我在学习 celery 方面取得了一些进展，我可以解雇工人并开始任务没有问题。在交互模式下使用它们非常直观。我很难找到我应该在我的 Flask 应用程序中存储异步结果的位置，以便以后检查它。我已更新问题以反映新状态
您收到的错误信息是什么？另一种选择是，在您的长时间运行的任务中，将任务的结果（或任务的状态）保存到数据库并从数据库中读取结果。

标签： python html flask celery

【解决方案1】：

我最终能够将我的任务 ID 保存在 Flask 的会话字典中

见下面的代码：

#!/usr/bin/env python

"""Page views."""

from flask import render_template, request
from flask import Flask

from celerytest import add

from time import sleep


app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def run():
    if request.method == 'GET':
        return render_template("template.html")
    else:
        form = request.form
        n1 = str(form.get("n1"))
        n2 = str(form.get("n2"))
        aysnc_res = add.delay(n1,n2)
        session['TASK_ID'] = async_res.id
        return render_template("loading.html")

@app.route('/loading')
def check_if_complete():
    aysnc_res = session['TASK_ID']
    if async_res.ready() == True:
        return render_template("template2.html", val=async_res.get())
    else:
        sleep(5)
        return render_template("loading.html")


if __name__ == '__main__':
    app.run()

【讨论】：

你可能不想在check_if_complete()“睡觉”。