Skip to content

Instantly share code, notes, and snippets.

@punchagan

punchagan/hello.py

Last active December 21, 2023 23:36
Show Gist options
  • Save punchagan/53600958c1799c2dcf26 to your computer and use it in GitHub Desktop.
Save punchagan/53600958c1799c2dcf26 to your computer and use it in GitHub Desktop.
Download ZIP
A simple Flask sockets example
Raw
hello.py
# Copy of http://stackoverflow.com/a/20104705
from flask import Flask, render_template
from flask_sockets import Sockets
app = Flask(__name__)
app.debug = True
sockets = Sockets(app)
@sockets.route('/echo')
def echo_socket(ws):
while True:
message = ws.receive()
ws.send(message[::-1])
@app.route('/')
def hello():
return 'Hello World!'
@app.route('/echo_test', methods=['GET'])
def echo_test():
return render_template('echo_test.html')
if __name__ == '__main__':
app.run()
Raw
templates-echo_test.html
<!DOCTYPE html>
<html>
<head>
<script type="text/javascript">
var ws = new WebSocket("ws://localhost:8000/echo");
ws.onopen = function() {
ws.send("socket open");
};
ws.onclose = function(evt) {
alert("socket closed");
};
ws.onmessage = function(evt) {
alert(evt.data);
};
</script>
</head>
</html>
@alexmclarty
Copy link

I've tried gunicorn and various other ways to make this work but I just get a 404.

@engelmav
Copy link

I've tried gunicorn and various other ways to make this work but I just get a 404.

Have you tried starting the flask app with gunicorn, using the parameter --worker-class flask_sockets.worker?

@engelmav
Copy link

Also, this is working for ping/pong, ws.handler.websocket.send_frame("HB", ws.OPCODE_PING)

    @sockets.route("/ws/stream")
    def websocket(ws):
        messages_backend.register(ws)
        while not ws.closed:
            message = ws.receive()
            LOG.debug(f"message: {message}")
            ws.handler.websocket.send_frame("HB", ws.OPCODE_PING)
            # Context switch while `MessagesBackend.start` is running in the background.
            gevent.sleep(1)

@davidhariri
Copy link

The flask_sockets project is now archived and unsupported. We @AdaSupport had issues running it with gunicorn. This snippet solved our problem:

# NOTE: https://stackoverflow.com/a/64002375/2114288
import grpc.experimental.gevent as grpc_gevent
grpc_gevent.init_gevent()

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment