شرح WebSockets في Django وFastAPI: بناء شات لحظي (Real-Time)

شرح WebSockets في Django وFastAPI: بناء شات لحظي (Real-Time)

في التطبيقات الحديثة، لم يعد طلب HTTP التقليدي كافيًا لتقديم تجربة تفاعلية وسريعة للمستخدم، خصوصًا في تطبيقات الشات، الإشعارات اللحظية، الألعاب متعددة اللاعبين، ولوحات التحكم الحية. هنا يأتي دور WebSockets مع أُطر مثل Django وFastAPI لتوفير اتصال ثنائي الاتجاه (Two-way) ومستمر بين العميل (Client) والخادم (Server).

في هذا الشرح سنتناول:

  • مفهوم WebSockets ولماذا نستخدمه.
  • إعداد Django Channels لدعم WebSockets.
  • إعداد WebSocket في FastAPI.
  • بناء مثال شات لحظي بسيط يعمل على كل من Django وFastAPI.
  • أفضل الممارسات والتكامل مع Redis لتوسيع التطبيق.

إذا كنت مهتمًا أكثر بخلفية Django Channels داخليًا، يمكنك مراجعة: Django Channels: كيف يعمل النظام داخليًا؟

ما هي WebSockets ولماذا نحتاجها؟

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

هذا مهم في:

  • تطبيقات الشات (Chat).
  • لوحات تحكم لحظية (Real-Time Dashboards).
  • إشعارات Live.
  • الألعاب أونلاين.

مع WebSockets، يصبح الخادم قادرًا على دفع البيانات للمستخدم دون أن يطلبها المستخدم صراحة في كل مرة.

نظرة عامة: Django FastAPI WebSockets

كل من Django وFastAPI يدعمان WebSockets، لكن بطريقة مختلفة:

  • Django يعتمد على Django Channels لتوفير طبقة ASGI تدعم WebSockets، بالإضافة إلى مفهوم Channels وGroups لإدارة الاتصالات.
  • FastAPI مبني من الأساس على ASGI، ويدعم WebSockets بشكل مباشر عبر WebSocket وWebSocketRoute مع إدارة بسيطة لاتصالات الشات.

إذا كنت تعمل أصلًا مع المصادقة والأمان في هذه الأطر، فقد يفيدك أيضًا: تنفيذ OAuth2 في Django و FastAPI: دليل عملي كامل.

إعداد WebSockets في Django باستخدام Django Channels

1. تثبيت الحزم المطلوبة

أول خطوة هي تثبيت channels في مشروع Django: 

pip install channels

في حال كنت تنوي دعم توزيع الاتصالات بين أكثر من سيرفر أو التكامل مع Redis، عادة ستحتاج: 

pip install channels_redis

2. تهيئة settings.py

في ملف settings.py قم بالتالي:

  1. إضافة channels إلى INSTALLED_APPS: 
    INSTALLED_APPS = [
    # ...
    "channels",
    # ...
]
  2. تعيين ASGI_APPLICATION: 
    ASGI_APPLICATION = "project_name.asgi.application"
  3. إعداد Channel Layer (اختياريًا مع Redis، لكنه مهم لتطبيق شات حقيقي): 
    CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

إذا أردت تعمقًا في استخدام Redis مع Django وFastAPI لتقليل الحمل على قواعد البيانات، راجع: Redis كمخزن مؤقت للتطبيقات.

3. إعداد ASGI في Django

أنشئ أو عدّل ملف asgi.py في جذر المشروع (نفس مكان settings.py): 

import os
import django
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_application
from django.urls import path
from chat.consumers import ChatConsumer

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project_name.settings")
django.setup()

django_asgi_app = get_asgi_application()

websocket_urlpatterns = [
    path("ws/chat/<room_name>/", ChatConsumer.as_asgi()),
]

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": AuthMiddlewareStack(
        URLRouter(websocket_urlpatterns)
    ),
})

هنا نفترض أن لدينا تطبيق اسمه chat يحتوي على ChatConsumer.

4. إنشاء Consumer للشات

في تطبيق chat أنشئ ملف consumers.py: 

from channels.generic.websocket import AsyncWebsocketConsumer
import json

class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"

        # الانضمام إلى مجموعة الغرفة
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # مغادرة مجموعة الغرفة
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # استقبال رسالة من WebSocket
    async def receive(self, text_data=None, bytes_data=None):
        data = json.loads(text_data)
        message = data["message"]

        # إرسال الرسالة إلى جميع الأعضاء في المجموعة
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                "type": "chat_message",
                "message": message,
            }
        )

    # استقبال رسالة من group_send
    async def chat_message(self, event):
        message = event["message"]

        # إرسال الرسالة مرة أخرى إلى WebSocket
        await self.send(text_data=json.dumps({
            "message": message
        }))

5. ربط المسارات (routing.py)

يمكنك أيضًا تنظيم مسارات WebSocket في ملف routing.py في تطبيق chat: 

from django.urls import re_path
from .consumers import ChatConsumer

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<room_name>\w+)/$", ChatConsumer.as_asgi()),
]

ثم تستورد websocket_urlpatterns في asgi.py بدل تعريفها هناك.

6. واجهة HTML بسيطة لدردشة Django

ملف HTML بسيط في templates/chat_room.html: 

<!DOCTYPE html>
<html lang="ar">
<head>
    <meta charset="UTF-8">
    <title>غرفة الشات</title>
</head>
<body>
    <h3>غرفة الشات: {{ room_name }}</h3>
    <ul id="messages"></ul>
    <input id="messageInput" type="text" placeholder="اكتب رسالة..." autocomplete="off">
    <button id="sendBtn">إرسال</button>

    <script>
        const roomName = "{{ room_name }}";
        const wsScheme = window.location.protocol === "https:" ? "wss" : "ws";
        const chatSocket = new WebSocket(
            wsScheme + "://" + window.location.host + "/ws/chat/" + roomName + "/"
        );

        chatSocket.onmessage = function(e) {
            const data = JSON.parse(e.data);
            const li = document.createElement("li");
            li.textContent = data.message;
            document.getElementById("messages").appendChild(li);
        };

        chatSocket.onclose = function(e) {
            console.error("انقطع الاتصال مع WebSocket");
        };

        document.getElementById("sendBtn").onclick = function(e) {
            const messageInput = document.getElementById("messageInput");
            const message = messageInput.value;
            chatSocket.send(JSON.stringify({ "message": message }));
            messageInput.value = "";
        };
    </script>
</body>
</html>

بهذا أصبح لدينا شات لحظي بسيط باستخدام Django و WebSockets عبر Django Channels.

إعداد WebSockets في FastAPI لبناء شات لحظي

FastAPI يدعم WebSockets مباشرة دون موديول إضافي، بشرط تشغيل التطبيق عبر خادم ASGI مثل uvicorn: 

pip install fastapi "uvicorn[standard]"

1. هيكل مشروع بسيط

ملف main.py: 

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()


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 broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)


manager = ConnectionManager()


@app.websocket("/ws/chat")
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(data)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

بهذا الشكل، أي عميل (متصفح أو تطبيق) يتصل على /ws/chat سيتم إضافته لقائمة الاتصالات النشطة، وأي رسالة يستقبلها الخادم سيتم بثها للجميع.

2. HTML بسيط لشات FastAPI

للتجربة السريعة، أنشئ ملف chat.html في نفس المجلد، أو قدّم صفحة عبر StaticFiles. مثال HTML: 

<!DOCTYPE html>
<html lang="ar">
<head>
    <meta charset="UTF-8">
    <title>شات FastAPI</title>
</head>
<body>
    <h3>شات FastAPI</h3>
    <ul id="messages"></ul>
    <input id="messageInput" type="text" placeholder="اكتب رسالة..." autocomplete="off">
    <button id="sendBtn">إرسال</button>

    <script>
        const wsScheme = window.location.protocol === "https:" ? "wss" : "ws";
        const chatSocket = new WebSocket(
            wsScheme + "://" + window.location.host + "/ws/chat"
        );

        chatSocket.onmessage = function(event) {
            const li = document.createElement("li");
            li.textContent = event.data;
            document.getElementById("messages").appendChild(li);
        };

        chatSocket.onclose = function(event) {
            console.error("تم إغلاق الاتصال");
        };

        document.getElementById("sendBtn").onclick = function() {
            const input = document.getElementById("messageInput");
            chatSocket.send(input.value);
            input.value = "";
        };
    </script>
</body>
</html>

شغّل التطبيق: 

uvicorn main:app --reload

ثم افتح http://127.0.0.1:8000/chat.html (إذا كنت تقدمه بشكل ثابت)، وافتح الصفحة في أكثر من تبويب لمشاهدة الرسائل تظهر لحظيًا.

لشروحات إضافية أعمق حول WebSockets في FastAPI، يمكنك مراجعة: التعامل مع WebSockets في FastAPI: تطبيق عملي.

بناء شات لحظي: مقارنة وتكامل بين Django وFastAPI

1. متى تستخدم Django Channels؟

  • إذا كان لديك تطبيق Django قائم (لوحة إدارة، ORM، نظام Auth).
  • تحتاج لنفس قواعد البيانات والنماذج (models) في تطبيق الشات.
  • تحتاج إلى حفظ الرسائل في قاعدة البيانات مع صلاحيات Django.

Django Channels يجعل WebSockets جزءًا طبيعيًا من مشروع Django، مع استخدام نفس مكونات النظام.

2. متى تفضّل FastAPI WebSockets؟

  • إذا كنت تريد خدمة مستقلة (Microservice) للشات.
  • إذا كنت بحاجة لأداء أعلى ومرونة أكبر في تصميم API.
  • إذا أردت بنيّة Async من البداية بدون قيود Django الكلاسيكية.

FastAPI مناسب لبناء WebSocket Gateway أو خدمة شات مستقلة يمكن استهلاكها من عدة تطبيقات (موبايل، ويب، ...).

3. مشاركة الحالة باستخدام Redis

لتوسيع شاتك إلى أكثر من عملية (workers) أو أكثر من سيرفر، ستحتاج لمخزن وسطي مشترك مثل Redis:

  • في Django Channels، ذلك يتم طبيعيًا عبر channels_redis كما أوضحنا في إعداد CHANNEL_LAYERS.
  • في FastAPI، يمكنك الاحتفاظ بالمستخدمين في Redis (IDs وRooms) واستخدام Pub/Sub لبث الرسائل.

حماية وأمان WebSockets في Django وFastAPI

تطبيق شات لحظي يعني اتصال دائم؛ لذا يجب التفكير في الأمان:

  • المصادقة (Auth): استخدام توكنات JWT، أو الجلسات (Sessions) في Django.
  • تحديد الصلاحيات: من يحق له دخول أي غرفة شات؟
  • تحديد المعدل (Rate Limiting): لمنع الرسائل المزعجة (Spam).
  • إغلاق الاتصال عند عدم النشاط: لتقليل استهلاك الموارد.

يمكن دمج نفس نظام التوثيق المستخدم في API (مثل OAuth2 أو JWT) مع WebSockets بحيث يتم تمرير التوكن إما في Query Params أو Header أثناء فتح الاتصال.

خلاصة: Django FastAPI WebSockets لشات لحظي فعّال

استخدام Django FastAPI WebSockets يفتح أمامك بابًا واسعًا لبناء تطبيقات لحظية (Real-Time) متقدمة:

  • Django + Channels مناسب إذا كان تطبيقك الكامل مبني على Django وتحتاج تكاملًا قويًا مع قاعدة البيانات ونظام الصلاحيات.
  • FastAPI يوفر طريقة مباشرة وخفيفة لبناء خدمات WebSockets عالية الأداء.
  • Redis يعتبر عنصرًا أساسيًا في توسيع التطبيق ودعم عدد كبير من الاتصالات المتزامنة.

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

حول المحتوى:

كيفية استخدام WebSockets للتواصل اللحظي، إعداد Django Channels وFastAPI WebSocket مع مثال عملي يعمل.

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

أضف تعليقك