我似乎无法弄清楚如何使用Flask的流媒体.这是我的代码:
@app.route('/scans/')
def scans_query():
url_for('static', filename='.*')
def generate():
yield render_template('scans.html')
for i in xrange(50):
sleep(.5)
yield render_template('scans.html', **locals())
return Response(stream_with_context(generate()))
在我的模板中:
<p>{% i %}</p>
我想在页面上看到一个每半秒钟改变一次的计数器.相反,我得到的最接近的是页面打印出下一行的每个数字.
jfs*_*jfs 57
要替换页面上的现有内容,您可能需要javascript,即,您可以发送它或使其为您发出请求,使用长轮询,websockets等.有很多方法可以做到这一点,这里有一个使用服务器发送事件:
#!/usr/bin/env python
import itertools
import time
from flask import Flask, Response, redirect, request, url_for
app = Flask(__name__)
@app.route('/')
def index():
if request.headers.get('accept') == 'text/event-stream':
def events():
for i, c in enumerate(itertools.cycle('\|/-')):
yield "data: %s %d\n\n" % (c, i)
time.sleep(.1) # an artificial delay
return Response(events(), content_type='text/event-stream')
return redirect(url_for('static', filename='index.html'))
if __name__ == "__main__":
app.run(host='localhost', port=23423)
地点static/index.html:
<!doctype html>
<title>Server Send Events Demo</title>
<style>
#data {
text-align: center;
}
</style>
<script src="http://code.jquery.com/jquery-latest.js"></script>
<script>
if (!!window.EventSource) {
var source = new EventSource('/');
source.onmessage = function(e) {
$("#data").text(e.data);
}
}
</script>
<div id="data">nothing received yet</div>
如果连接丢失,浏览器默认情况下会在3秒内重新连接.如果没有更多要发送的服务器可以返回404或只是发送一些'text/event-stream'内容类型以响应下一个请求.即使服务器有更多可以调用的数据,也要在客户端停止source.close().
注意:如果流不是无限的,那么使用其他技术(而不是SSE),例如,发送javascript片段来替换文本(无限<iframe>技术):
#!/usr/bin/env python
import time
from flask import Flask, Response
app = Flask(__name__)
@app.route('/')
def index():
def g():
yield """<!doctype html>
<title>Send javascript snippets demo</title>
<style>
#data {
text-align: center;
}
</style>
<script src="http://code.jquery.com/jquery-latest.js"></script>
<div id="data">nothing received yet</div>
"""
for i, c in enumerate("hello"):
yield """
<script>
$("#data").text("{i} {c}")
</script>
""".format(i=i, c=c)
time.sleep(1) # an artificial delay
return Response(g())
if __name__ == "__main__":
app.run(host='localhost', port=23423)
我在这里列出了html,以表明它没有更多的东西(没有魔法).这与上面相同,但使用模板:
#!/usr/bin/env python
import time
from flask import Flask, Response
app = Flask(__name__)
def stream_template(template_name, **context):
# http://flask.pocoo.org/docs/patterns/streaming/#streaming-from-templates
app.update_template_context(context)
t = app.jinja_env.get_template(template_name)
rv = t.stream(context)
# uncomment if you don't need immediate reaction
##rv.enable_buffering(5)
return rv
@app.route('/')
def index():
def g():
for i, c in enumerate("hello"*10):
time.sleep(.1) # an artificial delay
yield i, c
return Response(stream_template('index.html', data=g()))
if __name__ == "__main__":
app.run(host='localhost', port=23423)
地点templates/index.html:
<!doctype html>
<title>Send javascript with template demo</title>
<style>
#data {
text-align: center;
}
</style>
<script src="http://code.jquery.com/jquery-latest.js"></script>
<div id="data">nothing received yet</div>
{% for i, c in data: %}
<script>
$("#data").text("{{ i }} {{ c }}")
</script>
{% endfor %}
我想如果您要使用这样的模板,您可能需要使用stream_template此处给出的函数:http://flask.pocoo.org/docs/patterns/streaming/#streaming-from-templates
我没有测试这个,但它可能看起来像:
def stream_template(template_name, **context):
app.update_template_context(context)
t = app.jinja_env.get_template(template_name)
rv = t.stream(context)
rv.enable_buffering(5)
return rv
@app.route('/scans/')
def scans_query():
url_for('static', filename='.*')
def generate():
for i in xrange(50):
sleep(.5)
yield i
return Response(stream_template('scans.html', i=generate()))
| 归档时间： |
|
| 查看次数： |
37842 次 |
| 最近记录： |