flask_login 提供了一個方便的方式來管理用戶會話。當(dāng)你在 Flask 的 HTTP 視圖中使用它時，你可以簡單地使用 @login_required 裝飾器來確保用戶已登錄。

但是，flask_sockets 并沒有直接與 flask_login 集成。如果你想在建立 WebSocket 連接時檢查用戶是否已登錄，你需要采取一些額外的步驟。

以下是一個示例，說明如何在 flask_sockets 路由中使用 flask_login 進(jìn)行身份驗證：

1. 初始化 Flask、Flask-Login 和 Flask-Sockets：

from flask import Flask, request, session
from flask_login import LoginManager, current_user, UserMixin, login_required
from flask_sockets import Socketsapp = Flask(__name__)
app.config['SECRET_KEY'] = 'your-secret-key'
sockets = Sockets(app)
login_manager = LoginManager()
login_manager.init_app(app)

2. 定義用戶模型和加載用戶的回調(diào)：

class User(UserMixin):# For simplicity, this example does not use a real database.# Instead, it uses this dictionary to simulate user data.users = {"1": {"id": "1", "username": "user1", "password": "pass1"}}def __init__(self, id_, username):self.id = id_self.username = username@classmethoddef get(cls, id_):user_data = cls.users.get(id_)if not user_data:return Nonereturn User(id_=user_data["id"], username=user_data["username"])@login_manager.user_loader
def load_user(user_id):return User.get(user_id)

3. 定義 WebSocket 路由并檢查登錄狀態(tài)：

為了檢查用戶是否已登錄，我們會讀取 Flask 的會話數(shù)據(jù)，因為 flask_login 在其中存儲了有關(guān)用戶身份的信息。

@sockets.route('/socket')
def chat_socket(ws):# Flask's context might not be available in the WebSocket route,# so we manually load the user using Flask-Login's method.user = load_user(session.get("user_id"))# If user is not loaded (i.e., not logged in), we close the connection.if user is None:ws.close(reason="User not logged in.")returnwhile not ws.closed:message = ws.receive()if message:ws.send(f"Hello {user.username}, you said: {message}")

4. 創(chuàng)建一些基本的登錄和注銷路由：

from flask import render_template, redirect, url_for, request@app.route('/login', methods=["GET", "POST"])
def login():if request.method == "POST":username = request.form.get("username")password = request.form.get("password")user = [u for u in User.users.values() if u["username"] == username and u["password"] == password]if user:user = User.get(user[0]["id"])login_user(user)return redirect(url_for("index"))return render_template("login.html")@app.route('/logout')
@login_required
def logout():logout_user()return redirect(url_for("login"))@app.route('/')
@login_required
def index():return "Logged in as " + current_user.username

5. 運行應(yīng)用程序：

if __name__ == "__main__":from gevent.pywsgi import WSGIServerfrom geventwebsocket.handler import WebSocketHandlerserver = WSGIServer(('127.0.0.1', 5000), app, handler_class=WebSocketHandler)server.serve_forever()

注意：在上述代碼中，我們手動加載用戶并檢查他們是否已登錄，而不是使用 @login_required 裝飾器。在 WebSocket 路由中，由于 Flask 的請求上下文可能不可用，所以這是必要的。