در مدل Session-Based، کاربر ابتدا با HTTP لاگین می‌کند و یک session در سرور ایجاد می‌شود. سپس کلاینت می‌تواند WebSocket را باز کند و session برای احراز هویت استفاده می‌شود.

نمونه سرور FastAPI با session و WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, Response, Cookie
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import uuid

app = FastAPI()

# مدیریت session ساده با یک دیکشنری
sessions = {}

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.post("/login")
async def login(response: Response, username: str, password: str):
    # بررسی کاربری (در مثال ساده بدون DB)
    if username == "user" and password == "pass":
        session_id = str(uuid.uuid4())
        sessions[session_id] = {"username": username}
        response.set_cookie(key="session_id", value=session_id)
        return {"message": "Login successful"}
    return JSONResponse(status_code=401, content={"message": "Invalid credentials"})

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, session_id: str = Cookie(None)):
    if session_id is None or session_id not in sessions:
        await websocket.close(code=1008)
        return
    await websocket.accept()
    user = sessions[session_id]
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"{user['username']} sent: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

نمونه کلاینت پایتون با استفاده از session

import asyncio
import websockets

# فرض کنید مرورگر یا کلاینت session_id را دارد
SESSION_ID = "SESSION_ID_FROM_LOGIN"
uri = "ws://127.0.0.1:8000/ws"

async def websocket_client():
    async with websockets.connect(uri, extra_headers={"Cookie": f"session_id={SESSION_ID}"}) as websocket:
        print("Connected to server using session!")
        while True:
            message = input("Enter your message: ")
            await websocket.send(message)
            response = await websocket.recv()
            print(f"Server response: {response}")

asyncio.run(websocket_client())

توضیح:

• کاربر ابتدا با مسیر /login لاگین می‌کند و سرور یک session_id ایجاد و در کوکی ذخیره می‌کند.
• کلاینت هنگام اتصال WebSocket، session_id را در هدر Cookie ارسال می‌کند.
• سرور با بررسی session_id معتبر بودن اتصال را تایید می‌کند و ارتباط برقرار می‌شود.
• در صورت نامعتبر بودن session، اتصال WebSocket با کد 1008 بسته می‌شود.