47

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>

0.5 秒ごとに変化するカウンターをページに表示したいと思います。代わりに、私が得た最も近いものは、次の行に各番号を印刷するページです。

4

2 に答える 2

68

ページ上の既存のコンテンツを置き換えるには、JavaScriptが必要になる場合があります。つまり、JavaScriptを送信するか、リクエストを作成したり、長いポーリングやWebSocketなどを使用したりできます。これを行う方法はたくさんあります。サーバー送信イベントを使用する方法は次のとおりです。

#!/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秒後に再接続します。'text/event-stream'送信するものが他にない場合、サーバーは404を返すか、次の要求に応答してコンテンツタイプ以外のものを送信する可能性があります。サーバーにさらに多くのデータがある場合でもクライアント側で停止するには、を呼び出すことができます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 %}
于 2012-11-14T23:04:05.920 に答える
7

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()))
于 2012-11-14T20:53:10.343 に答える