التعامل مع WebSockets في FastAPI: تطبيق عملي

التعامل مع WebSockets في FastAPI: تطبيق عملي خطوة بخطوة

في الكثير من تطبيقات الويب الحديثة، لم يعد الطلب/الاستجابة التقليدي (Request/Response) كافيًا لتقديم تجربة تفاعلية وسريعة. هنا يأتي دور WebSockets مع FastAPI WebSockets لتوفير اتصال ثنائي الاتجاه بين العميل (المتصفح أو تطبيق الموبايل) والخادم في الوقت الحقيقي.

في هذا الشرح من افهم صح سنبني تطبيقًا عمليًا بسيطًا يعتمد على FastAPI WebSockets، ونتعرف على:

  • كيفية إنشاء WebSocket Endpoint في FastAPI
  • إدارة اتصالات متعددة
  • إنشاء غرف (Rooms) للدردشة أو التفاعل
  • التعامل مع مشاكل الأداء والموارد
  • أفضل الممارسات عند استخدام WebSockets في FastAPI

إذا كنت جديدًا على FastAPI، يمكنك مراجعة: بناء أول موقع باستخدام FastAPI و بناء RESTful APIs باستخدام FastAPI لتكوين صورة أوضح عن أساسيات الإطار.

ما هو WebSocket؟ ولماذا نحتاجه مع FastAPI؟

ببساطة، WebSocket هو بروتوكول يسمح بإنشاء اتصال مستمر بين العميل والخادم، بحيث يمكن لأي منهما إرسال بيانات في أي وقت بدون الحاجة لإعادة إنشاء اتصال جديد مع كل رسالة، بعكس HTTP التقليدي.

أمثلة على استخدام WebSockets:

  • تطبيقات الدردشة (Chat Apps)
  • لوحات التحكم (Dashboards) التي تعرض بيانات محدثة لحظيًا
  • تطبيقات الألعاب متعددة اللاعبين
  • التنبيهات الفورية والمباشرة (Notifications)

هنا يبرز دور FastAPI WebSockets

تهيئة مشروع FastAPI مع دعم WebSockets

نفترض أن بيئة بايثون جاهزة لديك. سنبدأ ببناء مشروع بسيط يدعم WebSockets.

1. تثبيت المتطلبات الأساسية

نثبّت FastAPI مع Uvicorn (سيرفر ASGI): 

pip install fastapi "uvicorn[standard]"

بعد التثبيت يمكنك إنشاء ملف مثلًا: main.py.

2. إنشاء تطبيق FastAPI أساسي

هيكل أولي للتطبيق:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "مرحبًا بك في تطبيق WebSockets باستخدام FastAPI"}

يمكنك تشغيل التطبيق: 

uvicorn main:app --reload

الآن ننتقل إلى إضافة دعم FastAPI WebSockets بشكل عملي.

إنشاء أول WebSocket Endpoint في FastAPI

لدعم WebSockets في FastAPI، نستخدم WebSocket و WebSocketDisconnect من الحزمة fastapi.

1. إضافة Endpoint لـ WebSocket

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"استقبلت: {data}")
    except WebSocketDisconnect:
        print("تم قطع اتصال العميل")

الشرح:

  • @app.websocket("/ws"): يعرّف مسار WebSocket على /ws.
  • await websocket.accept(): قبول الاتصال من العميل.
  • receive_text: استقبال رسالة نصية من العميل.
  • send_text: إرسال رد نصي للعميل.
  • WebSocketDisconnect: استثناء يتم رفعه عند قطع الاتصال من جهة العميل.

2. اختبار الاتصال عبر HTML بسيط

لنضيف صفحة HTML صغيرة لاختبار WebSocket:

from fastapi.responses import HTMLResponse

html = """
<!DOCTYPE html>
<html>
    <head>
        <title>اختبار FastAPI WebSockets</title>
    </head>
    <body>
        <h1>WebSocket Echo</h1>
        <input id="messageText" type="text" placeholder="اكتب رسالة" />
        <button onclick="sendMessage()">إرسال</button>
        <ul id="messages"></ul>
        <script>
            const ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                const messages = document.getElementById("messages");
                const li = document.createElement("li");
                li.innerText = event.data;
                messages.appendChild(li);
            }
            function sendMessage() {
                const input = document.getElementById("messageText");
                ws.send(input.value);
                input.value = "";
            }
        </script>
    </body>
</html>
"""

@app.get("/test", response_class=HTMLResponse)
async def get():
    return html

الآن بزيارة http://localhost:8000/test يمكنك كتابة رسالة، وسيقوم الخادم بإرجاعها (Echo) عبر WebSocket.

إدارة عدة اتصالات WebSocket في FastAPI

في التطبيقات الحقيقية، لا نتعامل مع اتصال واحد فقط، بل مع عدد كبير من المستخدمين المتصلين في نفس الوقت. نحتاج إلى آلية:

  • لتخزين اتصالات العملاء
  • لبث الرسائل للجميع أو لمجموعة محددة
  • لإزالة الاتصال عند انقطاعه

1. إنشاء مدير اتصالات (Connection Manager)

سنعرّف كلاس بسيط لإدارة الاتصالات:

from typing import List

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

هذا الكلاس يوفر لنا طريقة لإضافة اتصال جديد، إزالته، إرسال رسالة خاصة لمستخدم، وبث رسالة لجميع المتصلين.

2. تعديل WebSocket Endpoint لاستخدام ConnectionManager

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"مستخدم قال: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast("مستخدم غادر الدردشة")

الآن أي مستخدم يتصل على /ws/chat سيتمكن من إرسال رسائل تصل للجميع عبر broadcast.

إنشاء غرف (Rooms) باستخدام FastAPI WebSockets

في كثير من الأحيان لا نريد أن يتلقى جميع المستخدمين كل الرسائل، بل نريد تقسيمهم إلى غرف مختلفة (مثل غرف الدردشة أو غرف الاجتماعات).

1. تصميم بسيط لإدارة الغرف

سنعدل ConnectionManager ليدعم غرفًا متعددة، بحيث نربط كل غرفة بقائمة اتصالات خاصة بها. 

from typing import Dict, List

class RoomConnectionManager:
    def __init__(self):
        # room_id -> list of WebSockets
        self.rooms: Dict[str, List[WebSocket]] = {}

    async def connect(self, room: str, websocket: WebSocket):
        await websocket.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)

    def disconnect(self, room: str, websocket: WebSocket):
        if room in self.rooms:
            self.rooms[room].remove(websocket)
            if not self.rooms[room]:
                del self.rooms[room]

    async def send_to_room(self, room: str, message: str):
        if room in self.rooms:
            for connection in self.rooms[room]:
                await connection.send_text(message)

room_manager = RoomConnectionManager()

2. WebSocket Endpoint يدعم الغرف عبر Path Parameter

سنستخدم معرف الغرفة في مسار WebSocket:

@app.websocket("/ws/rooms/{room_id}")
async def websocket_room_endpoint(websocket: WebSocket, room_id: str):
    await room_manager.connect(room_id, websocket)
    await room_manager.send_to_room(room_id, f"مستخدم انضم إلى الغرفة {room_id}")
    try:
        while True:
            data = await websocket.receive_text()
            await room_manager.send_to_room(room_id, f"[{room_id}] قال: {data}")
    except WebSocketDisconnect:
        room_manager.disconnect(room_id, websocket)
        await room_manager.send_to_room(room_id, f"مستخدم غادر الغرفة {room_id}")

بهذه الطريقة، عند الاتصال على ws://localhost:8000/ws/rooms/room1 سيتم توجيه الرسائل إلى أعضاء الغرفة room1 فقط.

FastAPI WebSockets والأداء: مشاكل شائعة وحلول عملية

استخدام WebSockets مع FastAPI يوفر تجربة رائعة، لكن توجد تحديات متعلقة بالأداء، خاصة مع عدد كبير من الاتصالات.

1. مشكلة حظر التنفيذ (Blocking)

بما أن FastAPI مبني على async/await، يجب تجنّب أي عمليات متزامنة Blocking داخل دوال WebSocket (مثل الوصول لقاعدة بيانات بمكتبة متزامنة أو عمليات حسابية ثقيلة).

للتعمق في البرمجة غير المتزامنة يمكنك مراجعة: البرمجة غير المتزامنة في بايثون: تحسين الأداء باستخدام async و await.

حلول محتملة:

  • استخدام مكتبات متوافقة مع async للتعامل مع قواعد البيانات أو الخدمات الخارجية.
  • في حال الحاجة إلى عمليات ثقيلة، يمكن نقلها إلى Thread أو Process منفصل، كما شرحت في: Threading في بايثون.

2. استهلاك الذاكرة مع عدد كبير من الاتصالات

كل اتصال WebSocket يستهلك قدرًا من الذاكرة. عند آلاف الاتصالات قد يحدث ضغط على الخادم.

بعض النصائح:

  • تحديد أقصى عدد اتصالات (إذا كان ذلك منطقيًا لتطبيقك).
  • تنظيف الاتصالات غير النشطة: مثلاً، يمكنك تطبيق Ping/Pong أو Timeout لإغلاق الاتصالات التي لا ترسل أي بيانات لفترة طويلة.
  • استخدام أكثر من عامل (Workers) مع Uvicorn/Gunicorn لتوزيع الحمل.

3. Broadcast لعدد كبير من المستخدمين

تنفيذ حلقة for لإرسال الرسالة لكل اتصال في broadcast يمكن أن يصبح مكلفًا عند أعداد كبيرة من المستخدمين.

طرق تحسين:

  • تقليل عدد الرسائل المرسلة، مثل تجميع التحديثات وإرسالها دفعة واحدة كل فترة زمنية قصيرة.
  • استخدام Queue داخلية (مثل asyncio.Queue) لمعالجة الرسائل بشكل متسلسل وسلس.
  • الاستعانة بنظام Pub/Sub خارجي (Redis Pub/Sub مثلًا) إذا كان لديك عدة خوادم تحتاج لمزامنة الرسائل بينها.

التكامل بين FastAPI WebSockets و REST APIs

عادة لا يُستخدم WebSocket بمفرده، بل يكون مكملًا لـ REST API موجود بالفعل. مثال على سيناريو متكامل:

  • تسجيل المستخدم وتوثيق الجلسات عبر REST (JWT أو OAuth2).
  • اتصال WebSocket يستلم Token في Query أو Header للتحقق من هوية المستخدم.
  • ربط المستخدم بغرفة معينة بناءً على بياناته أو طلبه.

هذا الدمج يسمح لك بالاستفادة من قوة REST في العمليات التقليدية (CRUD) وقوة FastAPI WebSockets في التحديثات اللحظية.

نصائح وأفضل ممارسات عند استخدام FastAPI WebSockets

  • استخدم async/await في كل مكان داخل Endpoints الخاصة بـ WebSockets.
  • تعامل مع الأخطاء والاستثناءات داخل حلقات استقبال/إرسال الرسائل حتى لا يتوقف الخادم بشكل غير متوقع.
  • أمّن اتصال WebSocket:
    • عند نشر التطبيق استخدم wss:// بدل ws:// مع HTTPS.
    • تحقق من هوية المستخدم (Authentication) قبل قبول الاتصال أو عند الرسالة الأولى.
  • صمّم بروتوكول رسائل واضح:
    • يفضل استخدام JSON في الرسائل، مع حقول مثل type وpayload.
    • مثال: {"type": "message", "room": "room1", "text": "مرحبا"}.
  • راقب الأداء:
    • استخدم أدوات مراقبة (Monitoring) لمتابعة عدد الاتصالات، استهلاك الذاكرة، وزمن الاستجابة.
    • اختبر التطبيق تحت حمل (Load Testing) قبل النشر الفعلي.

مثال تطبيقي مبسط: تطبيق دردشة بغرف باستخدام FastAPI WebSockets

لنلخّص ما سبق في مثال واحد مبسط يدمج إدارة الغرف مع الرسائل: 

from typing import Dict, List

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse

app = FastAPI()

class RoomConnectionManager:
    def __init__(self):
        self.rooms: Dict[str, List[WebSocket]] = {}

    async def connect(self, room: str, websocket: WebSocket):
        await websocket.accept()
        if room not in self.rooms:
            self.rooms[room] = []
        self.rooms[room].append(websocket)

    def disconnect(self, room: str, websocket: WebSocket):
        if room in self.rooms:
            self.rooms[room].remove(websocket)
            if not self.rooms[room]:
                del self.rooms[room]

    async def send_to_room(self, room: str, message: str):
        if room in self.rooms:
            for connection in self.rooms[room]:
                await connection.send_text(message)

room_manager = RoomConnectionManager()

html = """
<!DOCTYPE html>
<html>
  <head><title>غرف دردشة FastAPI WebSockets</title></head>
  <body>
    <h1>غرفة دردشة</h1>
    <input id="room" placeholder="اسم الغرفة" />
    <button onclick="joinRoom()">انضمام للغرفة</button>
    <br/><br/>
    <input id="messageText" type="text" placeholder="اكتب رسالة" />
    <button onclick="sendMessage()">إرسال</button>
    <ul id="messages"></ul>
    <script>
      let ws;
      function joinRoom() {
        const room = document.getElementById("room").value;
        if (!room) return;
        ws = new WebSocket(`ws://localhost:8000/ws/rooms/${room}`);
        ws.onmessage = function(event) {
          const messages = document.getElementById("messages");
          const li = document.createElement("li");
          li.innerText = event.data;
          messages.appendChild(li);
        }
      }
      function sendMessage() {
        if (!ws) return;
        const input = document.getElementById("messageText");
        ws.send(input.value);
        input.value = "";
      }
    </script>
  </body>
</html>
"""

@app.get("/", response_class=HTMLResponse)
async def get():
    return html

@app.websocket("/ws/rooms/{room_id}")
async def websocket_room_endpoint(websocket: WebSocket, room_id: str):
    await room_manager.connect(room_id, websocket)
    await room_manager.send_to_room(room_id, f"مستخدم انضم إلى الغرفة {room_id}")
    try:
        while True:
            data = await websocket.receive_text()
            await room_manager.send_to_room(room_id, f"[{room_id}] قال: {data}")
    except WebSocketDisconnect:
        room_manager.disconnect(room_id, websocket)
        await room_manager.send_to_room(room_id, f"مستخدم غادر الغرفة {room_id}")

يمكنك تشغيل هذا المثال، فتح أكثر من تبويب في المتصفح، والانضمام إلى نفس الغرفة أو غرف مختلفة لمشاهدة كيفية عمل FastAPI WebSockets في تطبيق حي.

الخلاصة

استخدام FastAPI WebSockets يفتح أمامك بابًا لبناء تطبيقات تفاعلية في الوقت الحقيقي، مثل الدردشة، التحديثات اللحظية، الألعاب، ولوحات التحكم. في هذا المقال طبقنا:

  • إنشاء WebSocket Endpoint بسيط في FastAPI.
  • إدارة عدة اتصالات باستخدام ConnectionManager.
  • بناء غرف (Rooms) والتعامل مع الانضمام والمغادرة.
  • مناقشة مشاكل الأداء وأهم حلولها.

بعد فهم أساسيات WebSockets مع FastAPI، يمكنك الانتقال لمواضيع أوسع مثل توزيع الحمل، استخدام Redis Pub/Sub، أو مقارنة هذا النهج مع أطر أخرى مثل Django Channels، كما في شرح: بناء تطبيق دردشة باستخدام Django Channels.

بهذا تكون لديك قاعدة قوية للانطلاق في بناء أي تطبيق يعتمد على الاتصال ثنائي الاتجاه وتحديث البيانات في الوقت الحقيقي باستخدام FastAPI WebSockets.

حول المحتوى:

بناء تطبيق تفاعلي يستخدم WebSockets في FastAPI: الاتصال ثنائي الاتجاه، إدارة الغرف، التعامل مع اتصالات متعددة، ومشاكل الأداء والحلول.

هل كان هذا مفيدًا لك؟

أضف تعليقك