بناء واجهات برمجية REST عالية الأداء باستخدام FastAPI وPydantic لتطبيقات البيانات في الوقت الفعلي

تتطلب التطبيقات الويب الحديثة واجهات برمجية سريعة وموثوقة وقابلة للتوسع للتعامل مع معالجة البيانات في الوقت الفعلي. في هذا الدليل الشامل، سنتعرف على كيفية استغلال FastAPI وPydantic لبناء واجهات برمجية REST عالية الأداء يمكنها تقديم تطبيقات البيانات في الوقت الفعلي بكفاءة.

لماذا FastAPI وPydantic؟

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

البنية الأساسية

لنبدأ بفحص البنية الأساسية لتطبيق FastAPI. يوضح المثال التالي هيكلًا أساسيًا لواجهة برمجية بيانات في الوقت الفعلي:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import asyncio
from datetime import datetime

app = FastAPI(title="Real-Time Data API", version="1.0.0")

class DataPoint(BaseModel):
    id: int
    value: float
    timestamp: datetime
    source: str

class DataResponse(BaseModel):
    data: List[DataPoint]
    metadata: dict

@app.get("/data", response_model=DataResponse)
async def get_real_time_data():
    # محاكاة معالجة البيانات في الوقت الفعلي
    data_points = [
        DataPoint(
            id=1,
            value=42.5,
            timestamp=datetime.now(),
            source="sensor_001"
        )
    ]
    
    return DataResponse(
        data=data_points,
        metadata={"count": len(data_points), "processed_at": datetime.now()}
    )

تحسين الأداء

من أقوى ميزات FastAPI هي ميزات تحسين الأداء. إليك كيفية تنفيذ المعالجة غير المتزامنة للبيانات في الوقت الفعلي:

from fastapi import BackgroundTasks
import asyncio
from concurrent.futures import ThreadPoolExecutor

async def process_real_time_data(data: List[DataPoint]) -> List[DataPoint]:
    """محاكاة معالجة ثقيلة في الخلفية"""
    # في السيناريوهات الحقيقية، قد تتضمن هذه عمليات قواعد بيانات،
    # أو استنتاج التعلم الآلي، أو مكالمات واجهة برمجية خارجية
    await asyncio.sleep(0.1)  # محاكاة عمل غير متزامن
    return [point for point in data if point.value > 0]

@app.post("/data/batch", response_model=DataResponse)
async def process_batch_data(
    data: List[DataPoint], 
    background_tasks: BackgroundTasks
):
    # معالجة البيانات بشكل غير متزامن
    processed_data = await process_real_time_data(data)
    
    # إضافة إلى مهمة خلفية للمعالجة المستمرة
    background_tasks.add_task(process_background_tasks, processed_data)
    
    return DataResponse(
        data=processed_data,
        metadata={"count": len(processed_data), "processed_at": datetime.now()}
    )

async def process_background_tasks(data: List[DataPoint]):
    """التعامل مع مهام المعالجة الخلفية"""
    # تحديثات قاعدة البيانات، الإشعارات، أو تأثيرات جانبية أخرى
    pass

التحقق من صحة Pydantic لسلامة البيانات

يضمن التحقق من صحة مخطط Pydantic أن بيانات الوقت الفعلي تبقى متسقة وآمنة. إليك مثال متقدم مع التحقق المخصص:

from pydantic import validator, root_validator
from typing import Optional

class AdvancedDataPoint(BaseModel):
    id: int
    value: float
    timestamp: datetime
    source: str
    category: Optional[str] = None
    
    @validator('value')
    def value_must_be_positive(cls, v):
        if v < 0:
            raise ValueError('Value must be positive')
        return v
    
    @validator('source')
    def source_must_be_valid(cls, v):
        valid_sources = ['sensor_001', 'sensor_002', 'api_client']
        if v not in valid_sources:
            raise ValueError('Invalid source')
        return v
    
    @root_validator
    def validate_category_based_on_source(cls, values):
        source = values.get('source')
        category = values.get('category')
        
        if source == 'sensor_001' and category != 'temperature':
            raise ValueError('Temperature sensors must have temperature category')
        return values

class EventStream(BaseModel):
    events: List[AdvancedDataPoint]
    stream_id: str
    last_updated: datetime
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

التكامل مع WebSocket في الوقت الفعلي

لتطبيقات الوقت الفعلي الحقيقية، قم بدمج دعم WebSocket مع FastAPI:

from fastapi import WebSocket, WebSocketDisconnect
import json

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()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # معالجة البيانات في الوقت الفعلي
            response = {"message": f"Echo: {data}", "client": client_id}
            await manager.send_personal_message(json.dumps(response), websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

المراقبة والتحقق من الصحة

نفذ مراقبة شاملة لواجهة برمجية الوقت الفعلي الخاصة بك:

from fastapi.middleware.tracking import TrackingMiddleware

# إضافة وسطية التتبع
app.add_middleware(TrackingMiddleware)

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "services": {
            "database": "connected",
            "cache": "connected"
        }
    }

@app.get("/metrics")
async def get_metrics():
    import psutil
    return {
        "cpu_percent": psutil.cpu_percent(),
        "memory_percent": psutil.virtual_memory().percent,
        "active_connections": len(manager.active_connections)
    }

الاعتبارات عند النشر في البيئة الإنتاجية

للاستخدام في البيئة الإنتاجية، فكر في استخدام Gunicorn مع مهام uvicorn:

# gunicorn_config.py
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
timeout = 30
keepalive = 2
max_requests = 1000
max_requests_jitter = 100
preload = False

الخاتمة

يُعد بناء واجهات برمجية REST عالية الأداء لتطبيقات البيانات في الوقت الفعلي باستخدام FastAPI وPydantic مزيجًا قويًا من الميزات للتطوير. توفر الوثائق التلقائية لواجهة برمجية OpenAPI، والتحقق القوي من البيانات، والقدرات غير المتزامنة، هذه المجموعة مثالية للتطبيقات الحديثة. ومن خلال اتباع الأنماط الموضحة في هذا الدليل، ستتمكن من بناء واجهات برمجية لا تُضيع فقط في الأداء، بل تحافظ أيضًا على سلامة البيانات وتوفر تجربة ممتازة للمطورين من خلال وثائق شاملة لواجهة برمجية API.

سواء كنت تبني منصات بيانات الإنترنت من الأشياء، أو أنظمة التداول المالية، أو لوحة تحكم تحليلات الوقت الفعلي، فإن FastAPI وPydantic توفر الأساس اللازم لإنشاء واجهات برمجية قابلة للتوسع والصيانة، وعالية الأداء، قادرة على التعامل مع أحمال البيانات في الوقت الفعلي بكفاءة.