flask__websocket/websocket__flask-socketio_example/app_namespace.py - master - ilya.petrash/SimplePyScripts

flask__websocket/websocket__flask-socketio_example/app_namespace.py
135 строк4 KB
gil9red
Refactoring. Using black code style
12 май 2023, 17:34
12 май 2023, 17:344df3261
#!/usr/bin/env python

from threading import Lock

from flask import Flask, render_template, session, request
from flask_socketio import (
    SocketIO,
    Namespace,
    emit,
    join_room,
    leave_room,
    close_room,
    rooms,
    disconnect,
)


# Set this variable to "threading", "eventlet" or "gevent" to test the
# different async modes, or leave it set to None for the application to choose
# the best option based on installed packages.
async_mode = None

app = Flask(__name__)
app.config["SECRET_KEY"] = "secret!"
socketio = SocketIO(app, async_mode=async_mode)
thread = None
thread_lock = Lock()


def background_thread():
    """Example of how to send server generated events to clients."""
    count = 0
    while True:
        socketio.sleep(10)
        count += 1
        socketio.emit(
            "my_response",
            {"data": "Server generated event", "count": count},
            namespace="/test",
        )


@app.route("/")
def index():
    return render_template("index.html", async_mode=socketio.async_mode)


class MyNamespace(Namespace):
    def on_my_event(self, message):
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {"data": message["data"], "count": session["receive_count"]},
        )

    def on_my_broadcast_event(self, message):
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {"data": message["data"], "count": session["receive_count"]},
            broadcast=True,
        )

    def on_join(self, message):
        join_room(message["room"])
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {
                "data": "In rooms: " + ", ".join(rooms()),
                "count": session["receive_count"],
            },
        )

    def on_leave(self, message):
        leave_room(message["room"])
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {
                "data": "In rooms: " + ", ".join(rooms()),
                "count": session["receive_count"],
            },
        )

    def on_close_room(self, message):
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {
                "data": "Room " + message["room"] + " is closing.",
                "count": session["receive_count"],
            },
            room=message["room"],
        )
        close_room(message["room"])

    def on_my_room_event(self, message):
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {"data": message["data"], "count": session["receive_count"]},
            room=message["room"],
        )

    def on_disconnect_request(self):
        session["receive_count"] = session.get("receive_count", 0) + 1
        emit(
            "my_response",
            {"data": "Disconnected!", "count": session["receive_count"]},
        )
        disconnect()

    def on_my_ping(self):
        emit("my_pong")

    def on_connect(self):
        global thread
        with thread_lock:
            if thread is None:
                thread = socketio.start_background_task(background_thread)
        emit("my_response", {"data": "Connected", "count": 0})

    def on_disconnect(self):
        print("Client disconnected", request.sid)


socketio.on_namespace(MyNamespace("/test"))


if __name__ == "__main__":
    socketio.run(
        app,
        port=12000,
    )