The update() method when applied to the page, retrieves all open websockets associated with the specific WebPage instance and uses them to change the specific browser tab they connect with. wp.update() updates all browser tabs that were rendered using wp.,The WebPage instance wp is updated using the method update(). In order to understand what is happening, we need to make several distinctions.,The next line, shared_div.add_page(wp), is important. We need to add wp to the dictionary of pages shared_div is on. When we call the update method of shared_div, it will use this dictionary to update the appropriate browser tabs. ,JustPy uses this technology in order allow the server to "push" data to web pages. Here is an example of a program that implements a clock. Every second, the server pushes the updated time to all open web pages.
import justpy as jp
import time
import asyncio
wp = jp.WebPage(delete_flag = False)
clock_div = jp.Span(text = 'Loading...', classes = 'text-5xl m-1 p-1 bg-gray-300 font-mono', a = wp)
async def clock_counter():
while True:
clock_div.text = time.strftime("%a, %d %b %Y, %H:%M:%S", time.localtime())
jp.run_task(wp.update())
await asyncio.sleep(1)
async def clock_init():
jp.run_task(clock_counter())
async def clock_test():
return wp
jp.justpy(clock_test, startup = clock_init)import justpy as jp
wp = jp.QuasarPage(delete_flag = False)
editor = jp.QEditor(a = wp, kitchen_sink = True, height = '94vh')
def joint_edit():
return wp
jp.justpy(joint_edit)import justpy as jp
from datetime
import datetime
input_classes = 'm-2 bg-gray-200 font-mono appearance-none border-2 border-gray-200 rounded py-2 px-4 text-gray-700 focus:outline-none focus:bg-white focus:border-red-500'
button_classes = 'm-2 p-2 text-red-700 bg-white hover:bg-red-200 hover:text-red-500 border focus:border-red-500 focus:outline-none'
message_classes = 'ml-4 p-2 text-lg bg-red-500 text-white overflow-auto font-mono rounded-lg'
shared_div = jp.Div(classes = 'm-2 h-1/2 border overflow-auto', delete_flag = False)
header = jp.Div(text = 'Simple Message Board', classes = 'bg-red-100 border-l-4 border-red-500 text-red-700 p-4 text-3xl', delete_flag = False)
button_icon = jp.Icon(icon = 'paper-plane', classes = 'text-2xl', delete_flag = False)
button_text = jp.Div(text = 'Send', classes = 'text-sm', delete_flag = False)
message_icon = jp.Icon(icon = 'comments', classes = 'text-2xl text-green-600', delete_flag = False)
def message_initialize():
# Called once on startup
d = jp.Div(a = shared_div, classes = 'flex m-2 border')
time_stamp = jp.P(text = datetime.now().strftime('%Y-%m-%d %H:%M:%S'), classes = 'text-xs ml-2 flex-shrink-0')
p = jp.Pre(text = 'Welcome to the simple message board!', classes = message_classes)
d.add(message_icon, time_stamp, p)
async def send_message(self, msg):
if self.message.value:
d = jp.Div(classes = 'flex m-2 border')
time_stamp = jp.P(text = datetime.now().strftime('%Y-%m-%d %H:%M:%S'), classes = 'text-xs ml-2 flex-shrink-0')
p = jp.Pre(text = self.message.value, classes = message_classes)
d.add(message_icon, time_stamp, p)
shared_div.add_component(d, 0)
self.message.value = ''
# Clear message box after message is sent
await shared_div.update()
def message_demo():
wp = jp.WebPage()
outer_div = jp.Div(classes = 'flex flex-col h-screen', a = wp)
outer_div.add(header)
d = jp.Div(classes = 'flex', a = outer_div)
message = jp.Textarea(placeholder = 'Enter message here', a = d, classes = input_classes, debounce = 500)
send_button = jp.Button(a = d, click = send_message, classes = button_classes)
send_button.add(button_icon, button_text)
outer_div.add(shared_div)
shared_div.add_page(wp)
send_button.message = message
return wp
jp.justpy(message_demo, startup = message_initialize)import justpy as jp
import asyncio
button_classes = 'm-2 p-2 text-red-700 bg-white hover:bg-red-200 hover:text-red-500 border focus:border-red-500 focus:outline-none'
async def count_down(self, msg):
self.show = False
if hasattr(msg.page, 'd'):
msg.page.remove(msg.page.d)
bomb_icon = jp.Icon(icon = 'bomb', classes = 'inline-block text-5xl ml-1 mt-1', a = msg.page)
d = jp.Div(classes = 'text-center m-4 p-4 text-6xl text-red-600 bg-blue-500 faster', a = msg.page, animation = self.count_animation)
msg.page.d = d
for i in range(self.start_num, 0, -1):
d.text = str(i)
await msg.page.update()
await asyncio.sleep(1)
d.text = 'Boom!'
d.animation = 'zoomIn'
d.set_classes('text-red-500 bg-white')
bomb_icon.set_class('text-red-700')
self.show = True
def count_test(request):
start_num = int(request.query_params.get('num', 10))
animation = request.query_params.get('animation', 'flip')
wp = jp.WebPage()
count_button = jp.Button(text = 'Start Countdown', classes = button_classes, a = wp, click = count_down)
count_button.start_num = start_num
count_button.count_animation = animation
return wp
jp.justpy(count_test)This post is about creating Python Flask web pages that can be asynchronously updated by your Python Flask application at any point without any user interaction. We’ll be using Python Flask, and the Flask-SocketIO plug-in to achieve this. The working application is hosted on GitHub.,Running python 3.5.1 on windows 10. Made appropriate changes to application.py but no other files (should i?). index.html displays nicely but no numbers updated. Can you please help out?,For the purpose of updating the page once our user has first visited, we will be using Socket.io and the accomanying Flask addon built by the same Miguel Grinberg, Flask-Socketio (Miguel appears to be some sort of Python Flask God). Socket IO is a genius engine that allows real-time bidirectional event-based communication. Gone are the days of static HTML pages that load when you visit; with Socket technology, the server can continuously update your view with new information.,What I want to achieve here is a web page that is automatically updated for each user as a result of events that happened in the background on my server system. For example, allowing events like a continually updating message stream, a notification system, or a specific Twitter monitor / display. In this post, I show how to develop a bare-bones Python Flask application that updates connected clients with random numbers. Flask is an extremely lightweight and simple framework for building web applications using Python.
If you haven’t used Flask before, it’s amazingly simple, and to get started serving a very simple webpage only requires a few lines of Python:
# Basic Flask Python Web App
from flask
import Flask
app = Flask(__name__)
@app.route("/")
def hello():
return "Hello World!"
if __name__ == "__main__":
app.run()For Socket.io communication, “events” are triggered by either the server or connected clients, and corresponding callback functions are set to execute when these events are detected. Implementing event triggers or binding event callbacks are very simply implemented in Flask (after some initial setup) using:
# Basic blask server to
catch events
from flask
import Flask, render_template
from flask.ext.socketio
import SocketIO, emit
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
@socketio.on('my event') # Decorator to
catch an event called "my event":
def test_message(message): # test_message() is the event callback
function.
emit('my response', {
'data': 'got it!'
}) # Trigger a new event called "my response"
# that can be caught by another callback later in the program.
if __name__ == '__main__':
socketio.run(app)On the client side, a little bit of JavaScript wizardry with jQuery is used to handle incoming and trigger outgoing events. I would really recommend the JavaScript path on CodeSchool if you are not familiar with these technologies.
// Client Side Javascript to receive numbers.
$(document).ready(function(){
// start up the SocketIO connection to the server - the namespace 'test' is also included here if necessary
var socket = io.connect('http://' + document.domain + ':' + location.port + '/test');
// this is a callback that triggers when the "my response" event is emitted by the server.
socket.on('my response', function(msg) {
$('#log').append('<p>Received: ' + msg.data + '</p>');
});
//example of triggering an event on click of a form submit button
$('form#emit').submit(function(event) {
socket.emit('my event', {data: $('#emit_data').val()});
return false;
});
}); //Client-side Javascript code for handling random numbers
$(document).ready(function(){
//connect to the socket server.
var socket = io.connect('http://' + document.domain + ':' + location.port + '/test');
var numbers_received = [];
//receive details from server
socket.on('newnumber', function(msg) {
console.log("Received number" + msg.number);
//maintain a list of ten numbers
if (numbers_received.length >= 10){
numbers_received.shift()
}
numbers_received.push(msg.number);
numbers_string = '';
for (var i = 0; i < numbers_received.length; i++){
numbers_string = numbers_string + '<p>' + numbers_received[i].toString() + '</p>';
}
$('#log').html(numbers_string);
});
});And in the application.py file:
#Python code to start Asynchronous Number Generation
@app.route('/')
def index():
#only by sending this page first will the client be connected to the socketio instance
return render_template('index.html')
@socketio.on('connect', namespace = '/test')
def test_connect():
# need visibility of the global thread object
global thread
print('Client connected')
#Start the random number generator thread only
if the thread has not been started before.
if not thread.isAlive():
print("Starting Thread")
thread = RandomThread()
thread.start()Servers should check for errors in the headers at the time start_response is called, so that an error can be raised while the application is still running.,Note that this does not prevent server or framework developers from offering specialized services as custom variables inside the environ dictionary. This is the recommended approach for offering any such value-added services.,Note that a single object may play the role of a server with respect to some application(s), while also acting as an application with respect to some server(s). Such “middleware” components can perform such functions as:,and here is a snippet from a server/gateway that uses it to provide access to a platform-specific API:
HELLO_WORLD = b "Hello world!\n"
def simple_app(environ, start_response):
""
"Simplest possible application object"
""
status = '200 OK'
response_headers = [('Content-type', 'text/plain')]
start_response(status, response_headers)
return [HELLO_WORLD]
class AppClass:
""
"Produce the same output, but using a class
(Note: 'AppClass'
is the "application"
here, so calling it returns an instance of 'AppClass', which is then the iterable
return value of the "application callable"
as required by the spec.
If we wanted to use * instances * of 'AppClass'
as application objects instead, we would have to implement a '__call__'
method, which would be invoked to execute the application,
and we would need to create an instance
for use by the server or gateway.
""
"
def __init__(self, environ, start_response):
self.environ = environ self.start = start_response
def __iter__(self):
status = '200 OK'
response_headers = [('Content-type', 'text/plain')] self.start(status, response_headers) yield HELLO_WORLDimport os, sys
enc, esc = sys.getfilesystemencoding(), 'surrogateescape'
def unicode_to_wsgi(u):
# Convert an environment variable to a WSGI "bytes-as-unicode"
string
return u.encode(enc, esc).decode('iso-8859-1')
def wsgi_to_bytes(s):
return s.encode('iso-8859-1')
def run_with_cgi(application):
environ = {
k: unicode_to_wsgi(v) for k,
v in os.environ.items()
}
environ['wsgi.input'] = sys.stdin.buffer
environ['wsgi.errors'] = sys.stderr
environ['wsgi.version'] = (1, 0)
environ['wsgi.multithread'] = False
environ['wsgi.multiprocess'] = True
environ['wsgi.run_once'] = True
if environ.get('HTTPS', 'off') in('on', '1'):
environ['wsgi.url_scheme'] = 'https'
else:
environ['wsgi.url_scheme'] = 'http'
headers_set = []
headers_sent = []
def write(data):
out = sys.stdout.buffer
if not headers_set:
raise AssertionError("write() before start_response()")
elif not headers_sent:
# Before the first output, send the stored headers
status, response_headers = headers_sent[: ] = headers_set
out.write(wsgi_to_bytes('Status: %s\r\n' % status))
for header in response_headers:
out.write(wsgi_to_bytes('%s: %s\r\n' % header))
out.write(wsgi_to_bytes('\r\n'))
out.write(data)
out.flush()
def start_response(status, response_headers, exc_info = None):
if exc_info:
try:
if headers_sent:
# Re - raise original exception
if headers sent
raise exc_info[1].with_traceback(exc_info[2])
finally:
exc_info = None # avoid dangling circular ref
elif headers_set:
raise AssertionError("Headers already set!")
headers_set[: ] = [status, response_headers]
# Note: error checking on the headers should happen here,
# * after * the headers are set.That way,
if an error
# occurs, start_response can only be re - called with
# exc_info set.
return write
result = application(environ, start_response)
try:
for data in result:
if data: # don 't send headers until body appears
write(data)
if not headers_sent:
write('') # send headers now
if body was empty
finally:
if hasattr(result, 'close'):
result.close()from piglatin import piglatin class LatinIter: "" "Transform iterated output to piglatin, if it's okay to do so Note that the "okayness" can change until the application yields its first non - empty bytestring, so 'transform_ok' has to be a mutable truth value. "" " def __init__(self, result, transform_ok): if hasattr(result, 'close'): self.close = result.close self._next = iter(result).__next__ self.transform_ok = transform_ok def __iter__(self): return self def __next__(self): data = self._next() if self.transform_ok: return piglatin(data) # call must be byte - safe on Py3 else: return data class Latinator: # by default, don 't transform output transform = False def __init__(self, application): self.application = application def __call__(self, environ, start_response): transform_ok = [] def start_latin(status, response_headers, exc_info = None): # Reset ok flag, in case this is a repeat call del transform_ok[: ] for name, value in response_headers: if name.lower() == 'content-type' and value == 'text/plain': transform_ok.append(True) # Strip content - length if present, else it 'll be wrong response_headers = [(name, value) for name, value in response_headers if name.lower() != 'content-length' ] break write = start_response(status, response_headers, exc_info) if transform_ok: def write_latin(data): write(piglatin(data)) # call must be byte - safe on Py3 return write_latin else: return write return LatinIter(self.application(environ, start_latin), transform_ok) # Run foo_app under a Latinator 's control, using the example CGI gateway from foo_app import foo_app run_with_cgi(Latinator(foo_app))
raise exc_info[1].with_traceback(exc_info[2])def start_response(status, response_headers, exc_info = None): if exc_info: try: # do stuff w / exc_info here finally: exc_info = None # Avoid circular ref.
try:
# regular application code here
status = "200 Froody"
response_headers = [("content-type", "text/plain")]
start_response(status, response_headers)
return ["normal body goes here"]
except:
# XXX should trap runtime issues like MemoryError, KeyboardInterrupt
# in a separate handler before this bare 'except:'...
status = "500 Oops"
response_headers = [("content-type", "text/plain")]
start_response(status, response_headers, sys.exc_info())
return ["error body goes here"]If you modify your source code, running under WSGI, you need to restart the Apache server to reload the source code.,On the other hand, if you your WSGI process is run in so-called daemon mode (to verify, check if WSGIDaemonProcess is used in your apache configuration), you can reload the WSGI process by touching the .wsgi file.,In debug mode, the Flask app monitors your source code, and reload the source code if any modification is detected (i.e., auto-reloader). It also launches the debugger if an error is detected.,In the above examples, you need to restart the app if you make any modifications to the codes. For development, we can turn on the debug mode, which enables the auto-reloader as well as the debugger.
.wsgisys.path.append('/path/to/eclipse/plugins/org.python.pydev_x.x.x.x/pysrc')WSGIDaemonProcess myflaskapp user = flaskuser group = www - data threads = 5 python - path = /path/to / eclipse / plugins / org.python.pydev_x.x.x.x / pysrc/etc/environmentPYTHONPATH = '/path/to/eclipse/plugins/org.python.pydev_x.x.x.x/pysrc'Trueapp = Flask(__name__) app.debug = True # Enable reloader and debugger ...... if __name__ == '__main__': app.run()
app.run()app = Flask(__name__) ...... if __name__ == '__main__': app.run(debug = True) # Enable reloader and debugger
* Running on http: //127.0.0.1:5000/ (Press CTRL+C to quit)
*
Restarting with stat *
Debugger is active!
*
Debugger pin code: xxx - xxx - xxx
......