如何在异步线程内使用app.appcontext()方法来手动获取到上下文环境?

本人从https://blog.miguelgrinberg.com/post/dynamically-update-your-flask-web-pages-using-turbo-flask看到一个turbo-flask的扩展,应该是使用了websocket来实现的网页动态刷新(底层是trubo.js),打算用它来做一个服务器监控的小页面。
按照上面网址的内容,参照作者的代码,是初始化了一个app flask后,然后初始化turbo这个对象。
入口代码如下(具体jinja2模板的代码不再添加,可以看上面的链接)

import random
import re
import sys
from flask import Flask, render_template

app = Flask(__name__)

@app.route('/')
def index():
    return render_template('loadavg.html')

@app.context_processor
def inject_load():
    if sys.platform.startswith('linux'): 
        with open('/proc/loadavg', 'rt') as f:
            load = f.read().split()[0:3]
    else:
        load = [int(random.random() * 100) / 100 for _ in range(3)]
    return {'load1': load[0], 'load5': load[1], 'load15': load[2]}

然后,通过启动一个threading来实现刷新模板,代码如下:

@app.before_first_request
def before_first_request():
    threading.Thread(target=update_load).start()

# 注意此处的函数,是手动开启了上下文来实现的turbo实例调用push()方法。
def update_load():
    with app.app_context():
        while True:
            time.sleep(5)
            turbo.push(turbo.replace(render_template('loadavg.html'), 'load'))

本人试过,上述代码无任何问题。
但是我本着应用架构清晰的方式来打算用blueprint的方式将该扩展使用的时候,发现卡在了调用app_context()方法上,代码如下:

# 程序的包名为server_monitor,工厂函数涉及到turbo的代码
from flask import Flask
from server_monitor.extensions import turbo # turbo的实例
from server_monitor.utils import smart, ups, cpu

def create_app():

    app = Flask(__name__)

    app.register_blueprint(smart.bp)
    app.register_blueprint(ups.bp)
    app.register_blueprint(cpu.bp) # 需要调用turbo的blueprint

    app.config.from_mapping(
            HD_DEVICES = ['sd'+x for x in 'abcd']
            )

    # extensions
    turbo.init_app(app)  #注册turbo

    return app

app = create_app()

另外extensions.py文件

from turbo_flask import Turbo

turbo = Turbo()

上述代码中cpu.bp蓝本文件:

from flask import Blueprint, render_template
import sys
import threading
import time
from server_monitor.extensions import turbo


bp = Blueprint('cpu_load_avages', __name__, static_folder="../templates", url_prefix='/cpu')

@bp.route('/')
def index():
    return render_template('cpu.html')

@bp.context_processor
def inject_load():
    load = [0,0,0]
    if sys.platform.startswith('linux'):
        with open('/proc/loadavg', 'rt') as f:
            load = f.read().split()[0:3]

    return {'load1': load[0], 'load5': load[1], 'load15': load[2]}

@bp.before_app_first_request
def before_first_request():
    threading.Thread(target=update_load).start()

# 此处的函数应该调用app_context()上下文,但是如何调用呢?
def update_load():
    while True:
        time.sleep(5)
        turbo.push(turbo.replace(render_template('cpu.html'), 'load'))

上述代码的错误日志为:

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.7/threading.py", line 865, in run
    self._target(*self._args, **self._kwargs)
  File "/home/yang/flask_tutorial/server_monitor/utils/cpu.py", line 30, in update_load
    turbo.push(turbo.replace(render_template('cpu.html'), 'load'))
  File "/home/yang/.local/share/virtualenvs/flask_tutorial-fiBXDnIR/lib/python3.7/site-packages/flask/templating.py", line 146, in render_template
    ctx.app.update_template_context(context)
AttributeError: 'NoneType' object has no attribute 'app'

众所周知,想在一个非app装饰器函数中获取到上下文环境,可以使用类似如下方法:

with app.app_context():
    turbo.do_what() # 此处turbo为一个已经init_app(app)完毕的扩展实例

那么在blueprint中呢?如何实现?
看过书中,好像用到了一个_get_current_object()的方法,但是好像不能用了。

非常感谢辉大!为了flask这个产品开发的这个论坛!谢谢!本人也是第一次在这个论坛上上发言,flask刚接触不久,辉大的flask书写的非常详细!

请直接问你原始的问题(和 turbo-flask 相关),而不是自己推测的某种解决方案遇到的问题。另外请给出:

  • 你预期的行为
  • 实际的行为
  • 一个最小但是可以运行的示例程序

谢谢,已修正帖子内容。

书的第六章介绍过异步发送邮件函数的实现,和你的需求基本相同(注意导入 current_app 并把 59 行取消注释):

感谢,代码已经搞定了:

# ...
@bp.context_processor
def inject_load():
    load = [0,0,0]
    if sys.platform.startswith('linux'):
        with open('/proc/loadavg', 'rt') as f:
            load = f.read().split()[0:3]

    return {'load1': load[0], 'load5': load[1], 'load15': load[2]}

@bp.before_app_first_request
def before_first_request():
        threading.Thread(target=update_load, args=(current_app._get_current_object(),)).start()

def update_load(app):
    with app.app_context():
        while True:
            time.sleep(5)
            turbo.push(turbo.replace(render_template('cpu.html'), 'load'))

还有个附加问题:flask中的这个上下文到底是个什么样的对象,另外flask是如何同时处理多个不同请求的上下文的?

书里源码解析那一章谈到了这些话题。