البداية البسيطة (والساذجة)

أبسط نسخة تعمل فعلًا:

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text("echo: " + data)

دي «بتشتغل» — لكنها echo فقط، بلا حالة، والجلسات المنقطعة لا تُمسح. القصة الحقيقية تبدأ حين تكون الواجهة إشعاراتً وأحداثًا فعلية.

الحالة المشتركة ليست في ذاكرة عملية واحدة

مع أكثر من نسخة عامل (multi-worker)، قائمة الجلسات في ذاكرة عملية واحدة تضيع بالكامل. الحل: لوحة اتصالات مشتركة ثابتة الطبقة (ناشر عبر Redis pub/sub)، وكل عملية تسمع وتبث من خلالها. هذا امتداد «التزامن الأصلي أولًا» — إذاعة عبر أمر ذرّي واحد بدل تراسل يدوي بين العمليات.

إعادة الانضمام والجلسات الميتة

• الجلسة المنقطعة تترك كودًا معلقًا؛ والدورة الجديدة تطلب accept جديد.
• امسح الجلسات في finally مهما حدث:
  • await websocket.close()
  • احذفها من مدير الاتصالات — وإلا تتراكم قائمة موتى يستقبلون إشعارات في الفراغ.

التعريف بالعميل: token لا URL

لا تحدد الهوية من الرابط (/ws?user=x) — أي أحد يكتب الرابط يتقمص الآخر. المحميّ: تمرير token في أول رسالة أو في header مع فحصه قبل الانضمام؛ لو فشل → close(code=1008, reason="unauthorized").

البث الجماعي بأمان

عند حدثٍ جديد:

for conn in room.members():
    if conn.alive():
        await conn.send_text(json.dumps(event))

عندما تكبر الغرفة، تنقل البث لقناة مشتركة (Redis pub/sub)؛ ابدأ بسيطًا لكن اعرف نقطة التوسع مسبقًا.

اختبار WebSocket بلا عناء

مع TestClient المدمج في FastAPI:

with client.websocket_connect("/ws?token=valid") as ws:
    ws.send_text("ping")
    assert ws.receive_text() == "echo: ping"

الخلاصة

WebSocket ليس سحرًا: accept، اقرأ، ابثّت، واعرف مع من تتحدث. الذي يصنع الجودة — إدارة الحالة صراحةً في finally، إعادة انضمام مريحة، هوية مؤمّنة بالـ token، وتوسعة ذكية (pub/sub). ابدأ بأبسط سطر يعمل، وأضِف التعقيد واحدًا بعد آخر — لا العكس.