تطوير Backend لتعلّم الآلة باستخدام FastAPI

تعرّف على كيفية إنشاء واجهات APIs عالية الأداء لتعلّم الآلة باستخدام FastAPI.

الإعداد الأساسي

1. التثبيت

pip install fastapi uvicorn pydantic numpy pandas scikit-learn

2. هيكل المشروع

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn

app = FastAPI(
    title="API ML",
    description="API pour prédictions ML",
    version="1.0.0"
)

# نماذج البيانات
class PredictionRequest(BaseModel):
    features: List[float]
    model_type: Optional[str] = "default"

class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
    model_version: str

نقاط النهاية في API

1. المسارات الأساسية

@app.get("/")
async def root():
    return {"message": "API ML active"}

@app.get("/health")
async def health_check():
    return {"status": "healthy"}

@app.get("/models")
async def list_models():
    return {
        "available_models": [
            "classification",
            "regression",
            "clustering"
        ]
    }

2. نقاط نهاية التنبؤ

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        # تحميل النموذج
        model = load_model(request.model_type)
        
        # التنبؤ
        prediction = model.predict([request.features])[0]
        confidence = model.predict_proba([request.features])[0]
        
        return PredictionResponse(
            prediction=float(prediction),
            confidence=float(max(confidence)),
            model_version="1.0.0"
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

إدارة النماذج

1. التحميل والإدارة

from typing import Dict
import joblib
import os

class ModelManager:
    def __init__(self):
        self.models: Dict[str, object] = {}
        self.model_path = "models/"
        
    def load_model(self, model_type: str) -> object:
        if model_type not in self.models:
            path = os.path.join(self.model_path, f"{model_type}.pkl")
            if not os.path.exists(path):
                raise HTTPException(
                    status_code=404,
                    detail=f"Modèle {model_type} non trouvé"
                )
            self.models[model_type] = joblib.load(path)
        return self.models[model_type]

model_manager = ModelManager()

2. التحقق من صحة البيانات

from pydantic import validator
import numpy as np

class FeatureValidation(BaseModel):
    features: List[float]
    
    @validator('features')
    def validate_features(cls, v):
        if not v:
            raise ValueError("La liste des features ne peut pas être vide")
        if not all(isinstance(x, (int, float)) for x in v):
            raise ValueError("Toutes les features doivent être numériques")
        return v

الـ Middleware والأمان

1. Middleware مخصص

from fastapi import Request
import time

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

2. المصادقة

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token invalide"
        )

الاختبارات والتوثيق

1. اختبارات الوحدة

# test_api.py
from fastapi.testclient import TestClient
import pytest

client = TestClient(app)

def test_predict():
    response = client.post(
        "/predict",
        json={"features": [1.0, 2.0, 3.0]}
    )
    assert response.status_code == 200
    assert "prediction" in response.json()

2. توثيق API

# إضافة توثيق مخصص
@app.get("/docs")
async def custom_docs():
    return {
        "endpoints": {
            "/predict": {
                "description": "Endpoint de prédiction",
                "parameters": {
                    "features": "Liste de features numériques",
                    "model_type": "Type de modèle à utiliser"
                }
            }
        }
    }

أفضل الممارسات

1. الأداء

   • التخزين المؤقت للنماذج (Cache)
   • التحقق الفعّال من البيانات
   • إدارة الذاكرة
   • المراقبة (Monitoring)

2. الأمان

   • المصادقة
   • التحقق من صحة المدخلات
   • تحديد معدل الطلبات (Rate limiting)
   • تسجيل الأحداث (Logging)

3. الصيانة

   • الاختبارات الآلية
   • التوثيق
   • إدارة الإصدارات (Versioning)
   • المراقبة (Monitoring)

الخلاصة

النقاط الرئيسية التي يجب تذكّرها:

• واجهة API عالية الأداء
• أمان معزَّز
• توثيق شامل
• اختبارات آلية

التوصيات:

• اتّباع معايير REST
• تطبيق إجراءات الأمان
• الحفاظ على التوثيق محدَّثًا
• أتمتة الاختبارات
• التدرّب على FastAPI وتعميق المعرفة بها