@ app.route（'/ scans /'）
def scans_query（）：
url_for（'static'，filename ='。*'）
def generate（）：
产生render_template（'scans.html'）
for xrange（50）：
sleep（.5）
yield render_template（'scans.html'，** locals（））
返回响应（stream_with_context（generate（）））
< / pre>
< ; p> {％i％}< / p>
我想在页面上看到每半秒钟更换一次的计数器。相反，我得到的最接近的是打印出下一行每个数字的页面。
你可能需要JavaScript的页面，也就是说，你可以发送它或者让它为你发出请求，使用长轮询，websockets等。有很多方法可以做到这一点，下面是使用 其中 如果连接丢失，浏览器将在3秒内重新连接。如果没有什么更多的发送服务器可以返回404或只是发送一些不是 我已经在这里表示内部的html到它（没有魔法）。这里和上面一样，但使用模板： 其中 I can't seem to figure out how to using Flask's streaming. Here's my code: and in my template: I would like to see a counter on the page that changes every half second. Instead, the closest I've gotten is the page printing out each number on the next line. To replace existing content on the page you might need javascript i.e., you could send it or make it to make requests for you, use long polling, websockets, etc. There are many ways to do it, here's one that uses server send events: Where The browser reconnects by default in 3 seconds if the connection is lost. if there is nothing more to send the server could return 404 or just send some other than Note: if the stream is not meant to be infinite then use other techniques (not SSE) e.g., send javascript snippets to replace the text (infinite I've inlined the html here to show that there is nothing more to it (no magic). Here's the same as above but using templates: Where 这篇关于使用Python和Flask流数据的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！
＃！/ usr / bin / env python
导入itertools
导入时间$ b $ from烧瓶导入烧瓶，响应，重定向，请求，url_for
app = Flask（__ name__）
@ app.route（'/'）
def index（）：$ b $如果request.headers.get（'accept'）== 'text / event-stream'：
def events（）：
for i，c in enumerate（itertools.cycle（'\ | / - '））：
yielddata： ％s％d \\\
\\\
％（c，i）
time.sleep（.1）＃an a人工延迟
返回Response（events（），content_type ='text / event-stream'）
返回重定向（url_for（'static'，filename ='index.html'））
if __name__ ==__main__：
app.run（host ='localhost'，port = 23423）
static / index.html ：
$ b
<！doctype html>
< title>伺服器发送活动演示< / 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>尚未收到任何内容< / div>
文本/事件流内容类型响应下一个请求。要停止在客户端，即使服务器有更多的数据，你可以调用 source.close（）。注意：如果流不是无限的，那么使用其他技术（而不是SSE），例如，发送JavaScript片段来替换文本（无限< iframe> technique）：
$ b ＃！/ usr / bin / env python
导入时间$ b $ from烧瓶导入烧瓶，回应
$ b $ app =烧瓶（__ name__）
@app。 route（'/'）
def index（）：
def g（）：
yield<！doctype html>
< title> Send javascript snippets demo< ; / title>>
< style>
#data {
text-align：center;
}
< / style>
<脚本src =http://code.jquery.com/jquery-latest.js>< / script>
< div id =data>尚未收到任何内容< / div> $ b $ （hello）：
yield
< script>
$（＃data ）.text（{i} {c}）
< / script>
.format（i = i，c = c）
time.s leep（1）＃一个假的延迟
返回响应（g（））
if __name__ ==__main__：
app.run（host = localhost'，port = 23423）
$ b ＃！/ usr / bin / env python
从瓶子导入时间
瓶子，响应
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）
＃如果不需要立即反应，则取消注释
## rv.enable_buffering（5）
return rv
@ app.route（'/'）
def index（）：
def g（）：
for i，c in enumerate（hello* 10 ）：
time.sleep（.1）＃一个人工延迟
yield i，c
返回响应（stream_template（'index.html'，data = g（）））
if __name__ ==__main__：
app.run（host ='localhost'，port = 23423）
templates / index.html ：
<！doctype html>
< title>以模板演示方式发送JavaScript< / title>
< style>
#data {
text-align：center;
}
< / style>
< script src =http://code.jquery.com/jquery-latest.js>< / script>
< div id =data>尚未收到任何内容< / div>
{％for i，c in data：％}
< script>
$（＃data）。text（{{i}} {{c}}）
< / script>
{％endfor％}
@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>
解决方案 #!/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>
'text/event-stream' content type in response to the next request. To stop on the client side even if the server has more data you could call source.close(). <iframe> technique):#!/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)
#!/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 %}