Контекст Flask-socketio для Flask-SQLAlchemy

Я создаю приложение для создания отчетов в реальном времени, используя 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 flask-sqlalchemy flask-socketio
person Community    schedule 13.03.2018    source источник

Ответы (2)


arrow_upward
4
arrow_downward

Ваша проблема связана не с Flask-SocketIO, а с тем, что для использования Flask-SQLAlchemy в фоновом потоке вам нужен для этого контекст приложения.

Попробуйте следующее:

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(),))
person Miguel    schedule 14.03.2018
comment
Великолепно! Думал, это как-то связано с контекстом приложения! Извините за неправильно обозначенную ошибку. - person ; 14.03.2018

arrow_upward
3
arrow_downward

Ответ Мигеля правильный, но если в функцию передаются оба аргумента ключевых слов, возникает следующее исключение:

TypeError: background_thread() got an unexpected keyword argument 'args'

Так что вам нужно пройти только

thread = socketio.start_background_task(background_thread, (current_app._get_current_object()))
person j2logo    schedule 15.03.2018