Flask-SQLAlchemy的Flask-socketio上下文

问题描述

我正在使用Flask，Flask-Socketio和Flask-SQLAlchemy创建一个实时报告应用程序.我当前的设计在连接时创建一个后台线程，该线程查询API并将其插入到应用程序数据中.但是，运行此程序时，出现错误

RuntimeError: No application found. Either work inside a view function or push an application context.

flask_react_app.py:

from threading import Lock
from flask import Blueprint, render_template
from .model import Stock
from . import socketio

main = Blueprint('main', __name__)
thread_lock = Lock()
thread = None


@main.route('/')
def index():
    """Serve client-side application."""
    return render_template('index.html', async_mode=socketio.async_mode)


def generate_data():
    """
    returns the Stock object query set
    """
    return [i.serialize for i in Stock.query.all()]


def background_thread():
    """Example of how to send server generated events to clients."""
    while True:
        socketio.sleep(10)
        socketio.emit("my_response", generate_data())


@socketio.on("connect")
def test_connect():
    """
    Connect method which fires off thread to notify worker to get data.
    :return: emits initial data.
    """

    global thread
    with thread_lock:
        if thread is None:
            thread = socketio.start_background_task(target=background_thread)
    socketio.emit("my_response", generate_data())

我有两个问题.首先是提到的错误，其次，必须有更好的方法！

解决方案

您的问题与Flask-SocketIO不相关，但是与以下事实有关:要在后台线程中使用Flask-SQLAlchemy，您需要一个应用程序上下文. /p>

请尝试以下操作:

def background_thread(app):
    """Example of how to send server generated events to clients."""
    with app.app_context():
        while True:
            socketio.sleep(10)
            socketio.emit("my_response", generate_data())

然后在启动后台线程的地方将应用程序实例作为参数传递:

thread = socketio.start_background_task(target=background_thread, args=(current_app._get_current_object(),))

I am creating a live reporting application using Flask, Flask-Socketio and Flask-SQLAlchemy. My current design creates a background thread on connection which querys an API and inserts into the applications data. However, when running this, I get the error

RuntimeError: No application found. Either work inside a view function or push an application context.

flask_react_app.py:

from threading import Lock
from flask import Blueprint, render_template
from .model import Stock
from . import socketio

main = Blueprint('main', __name__)
thread_lock = Lock()
thread = None


@main.route('/')
def index():
    """Serve client-side application."""
    return render_template('index.html', async_mode=socketio.async_mode)


def generate_data():
    """
    returns the Stock object query set
    """
    return [i.serialize for i in Stock.query.all()]


def background_thread():
    """Example of how to send server generated events to clients."""
    while True:
        socketio.sleep(10)
        socketio.emit("my_response", generate_data())


@socketio.on("connect")
def test_connect():
    """
    Connect method which fires off thread to notify worker to get data.
    :return: emits initial data.
    """

    global thread
    with thread_lock:
        if thread is None:
            thread = socketio.start_background_task(target=background_thread)
    socketio.emit("my_response", generate_data())

I have two issues with this. Firstly the bug mentioned, secondly, there must be a better way!

解决方案

Your problem is not related to Flask-SocketIO, but to the fact that to use Flask-SQLAlchemy in a background thread you need an application context for it.

Try the following:

def background_thread(app):
    """Example of how to send server generated events to clients."""
    with app.app_context():
        while True:
            socketio.sleep(10)
            socketio.emit("my_response", generate_data())

Then in the place where you start the background thread pass the application instance as an argument:

thread = socketio.start_background_task(target=background_thread, args=(current_app._get_current_object(),))

这篇关于Flask-SQLAlchemy的Flask-socketio上下文的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！