在烧瓶 socketio 后台任务中访问应用程序上下文答案

【问题标题】：Accessing to application context in flask socketio background task在烧瓶 socketio 后台任务中访问应用程序上下文
【发布时间】：2018-03-05 14:52:40
【问题描述】：

我正在开发一个带有 Flask-SocketIO 库的 Flask 应用程序，用于在服务器后台任务和客户端之间进行实时通信。

我需要在后台任务中访问我的数据库，但我不知道如何实现它，因为没有初始化应用程序上下文。

我的代码是：

@socketio.on('connect', namespace='/rt/notifications/')
def start_notifications_thread():
    global thread
    with thread_lock:
        if thread is None:
            thread = socketio.start_background_task(target=notifications_job)

def notifications_job():
    last_alert_id = None
    while True:
        socketio.sleep(60)
        last_id = Alert.get_last_alert_id()  # Flask-SQLAlchemy model
        if last_alert_id is not None and last_id != last_alert_id:
            socketio.emit('new_alerts', {'msg': 'New alert', 'id': last_id}, namespace='/rt/notifications/')
        last_alert_id = last_id

【问题讨论】：

标签： python flask flask-socketio

【解决方案1】：

根据flask-socketio的作者miguelgrinberg的说法，以下是正确的做法。简单地说，我必须将我的应用程序实例传递给notifications_job 方法，如下所示：

from threading import Lock

from flask import current_app

from .. import socketio

thread = None
thread_lock = Lock()

def notifications_job(app):
    last_alert_id = None
    with app.app_context():
        while True:
            socketio.sleep(60)
            last_id = Alert.get_last_alert_id()  # Flask-SQLAlchemy   model
            if last_alert_id is not None and last_id != last_alert_id:
                socketio.emit('new_alerts', {'msg': 'New alert', 'id': last_id}, namespace='/rt/notifications/')
            last_alert_id = last_id

@socketio.on('connect', namespace='/rt/notifications/')
def start_notifications_thread():
    global thread
    with thread_lock:
        if thread is None:
            thread = socketio.start_background_task(notifications_job, current_app._get_current_object())

【讨论】：