I can't seem to figure out how to using Flask's streaming. Here's my code:
@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()))
and in my template:
<p>{% i %}</p>
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.
Flask is a lightweight Python web framework that provides useful tools and features for creating web applications in the Python Language. It gives developers flexibility and is an accessible framework for new developers because you can build a web application quickly using only a single Python file.
Streams are high-level async/await-ready primitives to work with network connections. Streams allow sending and receiving data without using callbacks or low-level protocols and transports.
This configuration contains "module": "flask", , which tells VS Code to run Python with -m flask when it starts the debugger. It also defines the FLASK_APP environment variable in the env property to identify the startup file, which is app.py by default, but allows you to easily specify a different file.
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:
#!/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)
Where 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>
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 '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()
.
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 <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)
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:
#!/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)
Where 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 %}
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With