ساخت APIهای REST با عملکرد بالا با FastAPI و Pydantic برای برنامه‌های داده‌های زنده

برنامه‌های وب مدرن نیازمند APIهای سریع، قابل اعتماد و مقیاس‌پذیر برای مدیریت پردازش داده‌های زنده هستند. در این راهنمای جامع، نحوه استفاده از FastAPI و Pydantic برای ساخت APIهای REST با عملکرد بالا که بتوانند به‌طور کارآمد برنامه‌های داده‌های زنده را پشتیبانی کنند را بررسی خواهیم کرد.

چرا FastAPI و Pydantic؟

FastAPI به عنوان یکی از قدرتمندترین فریم‌ورک‌های پایتون برای ساخت APIهای مدرن شناخته شده است که سرعت برنامه‌نویسی غیرهمزمان را با تولید خودکار مستندات API ترکیب می‌کند. Pydantic، کتابخانه همراه آن، قابلیت‌های اعتبارسنجی و سریال‌سازی داده‌های قوی را فراهم می‌کند که باعث می‌شود پاسخ‌های API شما هم قابل اعتماد و هم پیش‌بینی‌پذیر باشند.

معماری اصلی

بیایید با بررسی معماری اصلی یک برنامه FastAPI شروع کنیم. مثال زیر ساختاری پایه برای یک API داده‌های زنده را نشان می‌دهد:

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]:
    """شبیه‌سازی پردازش سنگین در پس‌زمینه"""
    # در سناریوهای واقعی، این می‌تواند شامل عملیات‌های پایگاه داده،
    # استنتاج هوش مصنوعی یا فراخوانی‌های API خارجی باشد
    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)

نظارت و چک‌های سلامت

برای API زنده خود نظارت جامع پیاده‌سازی کنید:

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

نتیجه‌گیری

ساخت APIهای REST با عملکرد بالا برای برنامه‌های داده‌های زنده با FastAPI و Pydantic، ترکیبی قدرتمند از ویژگی‌ها را برای توسعه‌دهندگان فراهم می‌کند. مستندات OpenAPI خودکار، اعتبارسنجی داده‌های قوی و قابلیت‌های غیرهمزمان باعث می‌شوند این پشته برای برنامه‌های مدرن مناسب باشد. با دنبال کردن الگوهای نشان داده شده در این راهنما، می‌توانید APIهایی بسازید که نه تنها عملکرد خوبی داشته باشند، بلکه سلامتی داده‌ها را نیز حفظ کنند و تجربه توسعه عالی را از طریق مستندات API جامع فراهم کنند.

چه در حال ساخت پلتفرم‌های داده IoT، سیستم‌های معاملات مالی یا پنل‌های تحلیل زنده باشید، FastAPI و Pydantic پایه‌ای را فراهم می‌کنند که برای ایجاد APIهای مقیاس‌پذیر، قابل نگهداری و با عملکرد بالا که بتوانند بار داده‌های زنده را به‌طور کارآمد مدیریت کنند، لازم است.