flask-socketio：入门

flask-socketio

入门

初始化

以下代码示例显示了如何将 Flask-SocketIO 添加到 Flask 应用程序：

from flask import Flask, render_template
from flask_socketio import SocketIO

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)

if __name__ == '__main__':
    socketio.run(app)

init_app()还支持初始化样式。要启动 Web 服务器，只需执行您的脚本。请注意 Web 服务器的启动方式。该 socketio.run()函数封装了web服务器的启动，替代了 app.run()标准的Flask开发服务器启动。当应用程序处于调试模式时，Werkzeug 开发服务器仍在内部使用和正确配置 socketio.run()。在生产模式下，如果可用，则使用 eventlet Web 服务器，否则使用 gevent Web 服务器。如果未安装 eventlet 和 gevent，则使用 Werkzeug 开发 Web 服务器。

Flask 0.11 中引入的命令可以用来启动基于 Werkzeug 的 Flask-SocketIO 开发服务器，但由于缺乏 WebSocket 支持，不推荐使用这种启动 Flask-SocketIO 服务器的方法。这个包的早期版本包括一个自定义版本的 命令，允许在 eventlet 和 gevent 生产服务器上使用 WebSocket，但是这个功能已经停止，有利于 上面显示的更健壮的启动方法。 flask run flask run socketio.run(app)

应用程序必须向加载 Socket.IO 库并建立连接的客户端提供一个页面：

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js" integrity="sha512-q/dWJ3kcmjBLU4Qc47E4A9kTB4m3wuTY7vkFJDTZKjTs8jhyGQnaUrxa0Ytd0ssMZhbNua9hE+E7Qv1j+DyZwA==" crossorigin="anonymous">script>
<script type="text/javascript" charset="utf-8">
    var socket = io();
    socket.on('connect', function() {
        socket.emit('my event', {data: 'I\'m connected!'});
    });
script>

接收消息

使用 SocketIO 时，消息作为事件被双方接收。在客户端使用 Javascript 回调。使用 Flask-SocketIO，服务器需要为这些事件注册处理程序，类似于视图函数处理路由的方式。

以下示例为未命名事件创建服务器端事件处理程序：

@socketio.on('message')
def handle_message(data):
    print('received message: ' + data)

上面的示例使用字符串消息。另一种类型的未命名事件使用 JSON 数据：

@socketio.on('json')
def handle_json(json):
    print('received json: ' + str(json))

最灵活的事件类型使用自定义事件名称。这些事件的消息数据可以是字符串、字节、int 或 JSON：

@socketio.on('my event')
def handle_my_custom_event(json):
    print('received json: ' + str(json))

自定义命名事件也可以支持多个参数：

@socketio.on('my_event')
def handle_my_custom_event(arg1, arg2, arg3):
    print('received args: ' + arg1 + arg2 + arg3)

当事件的名称是一个有效的 Python 标识符并且不与其他定义的符号冲突时， @socketio.event装饰器提供更紧凑的语法，从装饰函数中获取事件名称：

@socketio.event
def my_custom_event(arg1, arg2, arg3):
    print('received args: ' + arg1 + arg2 + arg3)

命名事件是最灵活的，因为它们不需要包含额外的元数据来描述消息类型。名称 message、 json、connect、 disconnect 和是保留的，不能用于命名事件。

Flask-SocketIO 还支持 SocketIO 命名空间，它允许客户端在同一个物理套接字上多路复用几个独立的连接：

@socketio.on('my event', namespace='/test')
def handle_my_custom_namespace_event(json):
    print('received json: ' + str(json))

如果未指定命名空间， '/'则使用具有该名称的默认全局命名空间。

对于装饰器语法不方便的情况， on_event可以使用该方法：

def my_function_handler(data):
    pass

socketio.on_event('my event', my_function_handler, namespace='/test')

客户端可能会请求一个确认回调，以确认收到他们发送的消息。从处理函数返回的任何值都将作为回调函数中的参数传递给客户端：

@socketio.on('my event')
def handle_my_custom_event(json):
    print('received json: ' + str(json))
    return 'one', 2

在上面的示例中，将使用两个参数调用客户端回调函数， 'one'并且 2. 如果处理函数没有返回任何值，则将调用客户端回调函数而不带参数。

发送消息

如上一节所示定义的 SocketIO 事件处理程序可以使用 send()和 emit() 函数向连接的客户端发送回复消息。

以下示例将收到的事件反弹回发送它们的客户端：

from flask_socketio import send, emit

@socketio.on('message')
def handle_message(message):
    send(message)

@socketio.on('json')
def handle_json(json):
    send(json, json=True)

@socketio.on('my event')
def handle_my_custom_event(json):
    emit('my response', json)

请注意如何 send()和 emit()分别用于未命名和命名事件。

使用命名空间时 send()， emit()默认使用传入消息的命名空间。可以使用可选 namespace参数指定不同的命名空间：

@socketio.on('message')
def handle_message(message):
    send(message, namespace='/chat')

@socketio.on('my event')
def handle_my_custom_event(json):
    emit('my response', json, namespace='/chat')

要发送带有多个参数的事件，请发送一个元组：

@socketio.on('my event')
def handle_my_custom_event(json):
    emit('my response', ('foo', 'bar', json), namespace='/chat')

SocketIO 支持确认消息已被客户端接收到的确认回调：

def ack():
    print('message was received!')

@socketio.on('my event')
def handle_my_custom_event(json):
    emit('my response', json, callback=ack)

使用回调时，Javascript 客户端会收到一个回调函数，以便在收到消息时调用。在客户端应用程序调用回调函数后，服务器调用相应的服务器端回调。如果使用参数调用客户端回调，则这些也作为参数提供给服务器端回调。

广播

SocketIO 的另一个非常有用的特性是消息的广播。Flask-SocketIO 通过and的 broadcast=True可选参数支持此功能： send() emit()

@socketio.on('my event')
def handle_my_custom_event(data):
    emit('my response', data, broadcast=True)

在启用广播选项的情况下发送消息时，连接到命名空间的所有客户端都会收到它，包括发送者。当不使用命名空间时，连接到全局命名空间的客户端会收到消息。请注意，不会为广播消息调用回调。

在此之前显示的所有示例中，服务器响应客户端发送的事件。但是对于某些应用程序，服务器需要是消息的发起者。这对于向客户端发送源自服务器的事件的通知非常有用，例如在后台线程中。和方法 socketio.send()可 socketio.emit()用于向所有连接的客户端广播：

def some_function():
    socketio.emit('some event', {'data': 42})

请注意 socketio.send()和 socketio.emit()与上下文感知 send()和的功能不同 emit()。另请注意，在上述用法中没有客户端上下文，因此 broadcast=True是假定的，不需要指定。

房间

对于许多应用程序，有必要将用户分组为可以一起处理的子集。最好的示例是具有多个房间的聊天应用程序，用户在其中接收来自他们所在的一个或多个房间的消息，而不是来自其他用户所在的其他房间的消息。 join_room()Flask-SocketIO 通过和 leave_room()函数支持这种房间概念：

from flask_socketio import join_room, leave_room

@socketio.on('join')
def on_join(data):
    username = data['username']
    room = data['room']
    join_room(room)
    send(username + ' has entered the room.', to=room)

@socketio.on('leave')
def on_leave(data):
    username = data['username']
    room = data['room']
    leave_room(room)
    send(username + ' has left the room.', to=room)

send()and emit()函数接受一个可选参数，该 to参数导致消息发送到给定房间中的所有客户端。

所有客户端在连接时都会分配一个房间，以连接的会话 ID 命名，可以从 request.sid. 给定的客户端可以加入任何房间，可以给它任何名称。当客户端断开连接时，它会从它所在的所有房间中删除。上下文无关 socketio.send() 和 socketio.emit()函数还接受一个 to参数以向房间中的所有客户端广播。

由于所有客户端都分配了一个个人房间，因此要将消息发送给单个客户端，可以使用客户端的会话 ID 作为 to参数。

连接事件

Flask-SocketIO 还分派连接和断开事件。以下示例显示了如何为它们注册处理程序：

@socketio.on('connect')
def test_connect(auth):
    emit('my response', {'data': 'Connected'})

@socketio.on('disconnect')
def test_disconnect():
    print('Client disconnected')

连接处理程序中的 auth参数是可选的。客户端可以使用它来传递身份验证数据，例如字典格式的令牌。如果客户端不提供身份验证详细信息，则此参数设置为 None。如果服务器定义了一个没有此参数的连接事件处理程序，那么 cient 传递的任何身份验证数据都将被丢弃。

连接事件处理程序可以返回 False以拒绝连接，也可以引发ConnectionRefusedError。这样就可以在此时对客户端进行身份验证。使用异常时，传递给异常的任何参数都会在错误包中返回给客户端。例子：

from flask_socketio import ConnectionRefusedError

@socketio.on('connect')
def connect():
    if not self.authenticate(request.args):
        raise ConnectionRefusedError('unauthorized!')

请注意，连接和断开连接事件是在每个使用的命名空间上单独发送的。

基于类的命名空间

作为上述基于装饰器的事件处理程序的替代方案，可以将属于命名空间的事件处理程序创建为类的方法。作为基flask_socketio.Namespace类提供以创建基于类的命名空间：

from flask_socketio import Namespace, emit

class MyCustomNamespace(Namespace):
    def on_connect(self):
        pass

    def on_disconnect(self):
        pass

    def on_my_event(self, data):
        emit('my_response', data)

socketio.on_namespace(MyCustomNamespace('/test'))

当使用基于类的命名空间时，服务器接收到的任何事件都被分派到一个名为事件名称的方法，该方法带有 on_前缀。例如，事件 my_event将由名为 的方法处理 on_my_event。如果接收到在命名空间类中没有定义相应方法的事件，则忽略该事件。基于类的命名空间中使用的所有事件名称必须使用在方法名称中合法的字符。

为了方便在基于类的命名空间中定义的方法，命名空间实例包括类中的几个方法的版本，这些方法在 未给出参数flask_socketio.SocketIO时默认为正确的命名空间。 namespace

如果事件在基于类的命名空间中具有处理程序，并且还具有基于装饰器的函数处理程序，则仅调用已装饰的函数处理程序。

错误处理

Flask-SocketIO 也可以处理异常：

@socketio.on_error()        # Handles the default namespace
def error_handler(e):
    pass

@socketio.on_error('/chat') # handles the '/chat' namespace
def error_handler_chat(e):
    pass

@socketio.on_error_default  # handles all namespaces without an explicit error handler
def default_error_handler(e):
    pass

错误处理函数将异常对象作为参数。

当前请求的消息和数据参数也可以使用 request.event变量进行检查，这对于事件处理程序之外的错误记录和调试很有用：

from flask import request

@socketio.on("my error event")
def on_my_event(data):
    raise RuntimeError()

@socketio.on_error_default
def default_error_handler(e):
    print(request.event["message"]) # "my error event"
    print(request.event["args"])    # (data,)

调试和故障排除

为了帮助您调试问题，可以将服务器配置为将日志输出到终端：

socketio = SocketIO(logger=True, engineio_logger=True)

该 logger参数控制与 Socket.IO 协议相关的日志记录，同时 engineio_logger控制源自低级 Engine.IO 传输的日志。这些参数可以设置为 True将日志输出到 stderr，或与 Python 的 logging包兼容的对象，日志应该发送到该对象。值 False禁用日志记录。

日志记录可以帮助确定连接问题、400 响应、性能不佳和其他问题的原因。

Original: https://blog.csdn.net/weixin_46020624/article/details/123619230
Author: Esun_nyy
Title: flask-socketio：入门