Flask Snippets

Snippets are unofficial and unmaintained.

This is an archived view of user-submitted snippets. Despite being hosted on the Flask site, they are not official. No Flask maintainer has curated or checked the snippets for security, correctness, or design.

Realtime server using the SSE protocol

Posted by Oskar Blom on 2013-08-01 @ 11:48 and filed in JavaScript

# author: oskar.blom@gmail.com
# Make sure your gevent version is >= 1.0
import gevent
from gevent.wsgi import WSGIServer
from gevent.queue import Queue

from flask import Flask, Response

import time

# SSE "protocol" is described here: http://mzl.la/UPFyxY
class ServerSentEvent(object):

    def __init__(self, data):
        self.data = data
        self.event = None
        self.id = None
        self.desc_map = {
            self.data : "data",
            self.event : "event",
            self.id : "id"

    def encode(self):
        if not self.data:
            return ""
        lines = ["%s: %s" % (v, k) 
                 for k, v in self.desc_map.iteritems() if k]
        return "%s\n\n" % "\n".join(lines)

app = Flask(__name__)
subscriptions = []

# Client code consumes like this.
def index():
    debug_template = """
         <h1>Server sent events</h1>
         <div id="event"></div>
         <script type="text/javascript">

         var eventOutputContainer = document.getElementById("event");
         var evtSrc = new EventSource("/subscribe");

         evtSrc.onmessage = function(e) {
             eventOutputContainer.innerHTML = e.data;


def debug():
    return "Currently %d subscriptions" % len(subscriptions)

def publish():
    #Dummy data - pick up from request for real data
    def notify():
        msg = str(time.time())
        for sub in subscriptions[:]:
    return "OK"

def subscribe():
    def gen():
        q = Queue()
            while True:
                result = q.get()
                ev = ServerSentEvent(str(result))
                yield ev.encode()
        except GeneratorExit: # Or maybe use flask signals

    return Response(gen(), mimetype="text/event-stream")

if __name__ == "__main__":
    app.debug = True
    server = WSGIServer(("", 5000), app)
    # Then visit http://localhost:5000 to subscribe 
    # and send messages by visiting http://localhost:5000/publish

This snippet by Oskar Blom can be used freely for anything you like. Consider it public domain.