Flask Celery socketio RuntimeError:在请求上下文之外工作

Question

我当前正在尝试设计一个烧瓶应用程序，该应用程序允许使用套接字和芹菜进行实时图表。我希望能够异步获取数据，然后通过套接字将其发送给客户端。但是，我会遇到错误：RuntimeError: Working outside of request context.我在插座首次连接时得到了。

堆栈跟踪

Traceback (most recent call last):
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/celery/app/trace.py", line 374, in trace_task
    R = retval = fun(*args, **kwargs)
  File "/Users/bev/PycharmProjects/flask_project/celery_config.py", line 15, in __call__
    return TaskBase.__call__(self, *args, **kwargs)
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/celery/app/trace.py", line 629, in __protected_call__
    return self.run(*args, **kwargs)
  File "/Users/bev/PycharmProjects/flask_project/main.py", line 20, in async_data
    send(jsonify({"result": sample(range(101), 6)}))
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/flask/json.py", line 251, in jsonify
    if current_app.config['JSONIFY_PRETTYPRINT_REGULAR'] and not request.is_xhr:
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/werkzeug/local.py", line 347, in __getattr__
    return getattr(self._get_current_object(), name)
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/werkzeug/local.py", line 306, in _get_current_object
    return self.__local()
  File "/Users/bev/.virtualenvs/flask_project/lib/python3.6/site-packages/flask/globals.py", line 37, in _lookup_req_object
    raise RuntimeError(_request_ctx_err_msg)
RuntimeError: Working outside of request context.
This typically means that you attempted to use functionality that needed
an active HTTP request.  Consult the documentation on testing for
information about how to avoid this problem.

main.py

from flask import Flask, render_template, jsonify
from flask_socketio import SocketIO, send
from random import sample
from celery_config import make_celery

app = Flask(__name__)
app.config["SECRET_KEY"] = "thisisasecret"
socketio = SocketIO(app)
app.config.update(
    CELERY_BROKER_URL="amqp://localhost//",
    CELERY_RESULT_BACKEND="rpc://"
)
celery = make_celery(app)

@celery.task(name="main.async_data")
def async_data():
    # for now this is very small as an example
    # preferably batched to be done every 15 minutes.
    send(jsonify({"result": sample(range(101), 6)}))
    return True

@app.route("/")
def index():
    return render_template("chart.html")

@socketio.on("connect")
def handle_connection():
    async_data.delay()
    print("You are connected and we are getting your data")

if __name__ == "__main__":
    socketio.run(app, debug=True)

celery_config.py

from celery import Celery

def make_celery(app):
    celery = Celery(app.import_name, backend=app.config['CELERY_RESULT_BACKEND'],
                    broker=app.config['CELERY_BROKER_URL'])
    celery.conf.update(app.config)
    TaskBase = celery.Task
    class ContextTask(TaskBase):
        abstract = True
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return TaskBase.__call__(self, *args, **kwargs)
    celery.Task = ContextTask
    return celery

图表JavaScript

let chartConfig = {
    type: "line",
    data: {
        labels: ["Jan", "Feb", "Mar", "Apr", "May", "Jun"],
        datasets: [{
            label: "GOOG",
            data: [],
            borderColor: "rgba(22, 172, 65, 1)",
            borderWidth: 1,
            fill:false
        }]
    }
};
let socket = io.connect("http://" + document.domain + ':' + location.port);
socket.on("connect", function() {
    socket.send("Connected Socket off to get data");
});
socket.on("message", function (data) {
    chartConfig.data.datasets[0].data = data.result;
    let ctx = document.getElementById("myChart").getContext("2d");
    let myLineChart = new Chart(ctx, chartConfig)
});

Answer 1

原因仅在烧瓶中。让我们进行一些测试（main.py）：

from random import sample
from flask import Flask, jsonify
from celery_config import make_celery
app = Flask(__name__)
app.config["SECRET_KEY"] = "thisisasecret"
app.config.update(
    CELERY_BROKER_URL="redis://localhost:6379/0", # or your broker
    CELERY_RESULT_BACKEND="redis://localhost:6379/0" # or your broker
)
celery = make_celery(app)
@celery.task(name="main.async_data")
def async_data():
    jsonify({"result": sample(range(101), 6)})
    return True
@app.route("/")
def index():
    async_data.delay()
    return 'test'
if __name__ == "__main__":
    app.run(debug=True)

运行芹菜和烧瓶应用程序，打开'/'路线。您将看到下一个错误：

RuntimeError：在请求上下文之外工作

它的工作原理。您使用jsonify方法。正如您在文档中看到的那样，它将JSON输出转变为具有应用程序/JSON MIMETYPE的响应对象。但是Celery task对烧瓶的当前/主动响应一无所知。这只是异步代码/处理。如果您需要在芹菜中与JSON合作，则可以使用任何LIB（JSON，UJSON，SIMPLEJSON等）。

现在，让我们将jsonify更改为json.dumps({"result": sample(range(101), 6)})。您会看到一切正常。

希望这会有所帮助。