本人从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书写的非常详细!