في المشهد الحديث لهندسة البرمجيات، تحولت الحاجة إلى التواصل الفوري من ميزة "اختيارية" إلى مطلب أساسي. سواء كانت لوحات تداول الأسهم المباشرة، أو أدوات التحرير التعاونية، أو تطبيقات الألعاب عالية التردد، يجب أن تتفاعل الأنظمة الآن فورًا مع تغييرات الحالة. إن استعلام HTTP التقليدي غير فعال في هذه السيناريوهات، حيث يؤدي إلى تأخير غير ضروري وزيادة الحمل على الخادم. هنا يبرز FastAPI، عند دمجه مع بروتوكولات WebSockets، كخيار رائد لبناء واجهات برمجة تطبيقات قوية وقائمة على الأحداث في Python.
تكمن قوة FastAPI في إطارها غير المتزامن (Asynchronous)، المبنية فوق Starlette و Pydantic. بينما تشتهر بوحدات REST عالية الأداء عبر HTTP، فإن قوتها الحقيقية في مجال الأحداث القائمة تُفتح من خلال WebSockets. من خلال الاستفادة من نظام `asyncio` في Python، يمكننا إنشاء اتصالات دائمة وثنائية الاتجاه تسمح للخادم بدفع البيانات إلى العملاء في اللحظة التي يحدث فيها الحدث، مما يلغي الحاجة إلى استعلام العميل بشكل متكرر للحصول على التحديثات.
فهم البنية: الأولوية للأداء غير المتزامن
يعتمد أساس أي تطبيق فوري على قدرته على التعامل مع اتصالات متزامنة دون تعطيل الخيط الرئيسي. في العالم المتزامن، يمكن لاتصال WebSocket طويل الأمد واحد أن يعطل الخادم بأكمله إذا لم تتم إدارته بعناية. يحل FastAPI هذه المشكلة بشكل أصلي. من خلال استخدام مفاتيح `async` و `await`، يمكن للخادم تضخيم آلاف اتصالات WebSocket المتزامنة على خيط واحد، والانتظار لأحداث الإدخال والإخراج (مثل الرسائل الواردة أو استعلامات قاعدة البيانات) دون إيقاف التنفيذ.
عند تصميم واجهة برمجة تطبيقات قائمة على الأحداث، تتضمن العملية عادةً ثلاثة مكونات: العميل، والوسيط/الخادم، وحافلة الأحداث. في إعداد FastAPI النقي، غالبًا ما نستخدم قاموسًا مشتركًا في الذاكرة لحفظ الأحداث أو نمجّج المنتج-المستهلك باستخدام طوابير `asyncio` لإدارة الحالة عبر العملاء المتصلين.
تنفيذ نقطة نهاية WebSocket
للبدء، نحدد مسار WebSocket باستخدام مُعدّل `@app.websocket`. على عكس المسارات القياسية لـ REST، تتعامل WebSockets مع المصافحة، وتقيم اتصالًا دائمًا، ثم تدخل في حلقة حيث يمكن للخادم قراءة وكتابة الرسائل إلى ما لا نهاية.
فيما يلي مثال عملي لنقطة نهاية 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()
# معالجة الرسالة أو triggering حدث هنا
await websocket.send_text(f"تم استلام النص: {data}")
except WebSocketDisconnect:
print("انقطع اتصال العميل")التوسع لعدة عملاء: نمط البث
مثال الإصداء أعلاه بسيط. تظهر القوة الحقيقية لواجهات برمجة التطبيقات القائمة على الأحداث عندما نرغب في بث حدث واحد إلى جميع العملاء المتصلين. لتحقيق ذلك، يجب علينا الحفاظ على سجل لاتصالات WebSocket النشطة. في بيئة الإنتاج، سيكون هذا غالبًا قناة نشر/اشتراك في Redis، ولكن للتعلم المتوسط، تعمل قائمة في الذاكرة بشكل مثالي.
لنقم بتحسين تطبيقنا لقبول الإشعارات من عميل واحد وبثها إلى الآخرين.
from typing import List
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
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")
async def websocket_endpoint(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("انقطع عميل")دمج الأحداث الخارجية والمهام الخلفية
في بنية الأحداث الحقيقية، غالبًا ما تنشأ الأحداث خارج طبقة WebSocket نفسها. قد يكون هذا تغييرًا في قاعدة البيانات، أو رسالة من واجهة برمجة تطبيقات خارجية، أو مهمة مجدولة. يسمح FastAPI لنا بتشغيل مهام خلفية باستخدام `BackgroundTasks` أو نمط `asyncio.create_task` للاستماع إلى هذه الأحداث الخارجية ودفعها عبر اتصالاتنا عبر WebSocket.
على سبيل المثال، يمكنك إنشاء خدمة خلفية تستعلم عن واجهة برمجة تطبيقات طقس طرف ثالث كل دقيقة وبث البيانات الجديدة إلى جميع عملاء لوحة المعلومات المتصلين. يفصل هذا بين توليد الحدث وتسليمه للعميل، مما يلتزم بمبدأ فصل الاهتمامات.
الخاتمة
يوفر بناء واجهات برمجة تطبيقات فورية وقائمة على الأحداث باستخدام FastAPI و WebSockets حلاً قويًا وقابلًا للتوسع لتطبيقات الويب الحديثة. من خلال الاستفادة من القدرات غير المتزامنة في Python، يمكن للمطورين إنشاء أنظمة استجابة وكفاءة وقدرة على التعامل مع أحمال عمل عالية التوازي. سواء كنت تبني تطبيق دردشة، أو نظام إشعارات مباشر، أو لوحة مراقبة معقدة، فإن مزيج أداء FastAPI واتصال WebSockets الدائم يقدم الأساس التقني المثالي. مع تقدمك في مشاريعك، تذكر مراعاة قابلية التوسع؛ بينما تعمل القوائم في الذاكرة للتطوير، فإن الانتقال إلى Redis لإدارة الحالة سيكون أمرًا حاسمًا لنشر الإنتاج.