文档:
PyPI: https://pypi.org/project/Flask-SocketIO/
Github: https://github.com/miguelgrinberg/Flask-SocketIO
doc: https://flask-socketio.readthedocs.io
socket.io: https://socket.io/
安装
pip install flask-socketio gevent-websocket
代码实例
from flask import Flask, render_template, request from flask_socketio import SocketIO app = Flask(__name__) app.config['SECRET_KEY'] = 'secret!' app.jinja_env.auto_reload = True socketio = SocketIO(app) @app.route('/') def index(): return render_template('index.html') @app.route('/send_all') def send_all(): """ 广播 :return: """ message = request.args.get('message') socketio.send(message) return {'status': 'ok'} @app.route('/send') def send_message(): """ 单独发送 :return: """ sid = request.args.get('sid') message = request.args.get('message') socketio.send(message, to=sid) return {'status': 'ok'} @socketio.on('connect') def connect(): print('connect') socketio.send({'sid': request.sid}) @socketio.on('disconnect') def disconnect(): print('disconnect') @socketio.on('message') def handle_message(data): print(data) socketio.send({'data': data}) if __name__ == '__main__': socketio.run(app, debug=True)
Http 发送测试请求
import requests params = { 'sid': 'm4AymrH2TIFHCcQNAAAF', 'message': 'send' } res = requests.get('http://localhost:5000/send', params=params) print(res.text) params = { 'message': 'send_all' } res = requests.get('http://localhost:5000/send_all', params=params) print(res.text)