Modern web uygulamaları, gerçek zamanlı veri işleme için hızlı, güvenilir ve ölçeklenebilir API'ler talep eder. Bu kapsamlı kılavuzda, yüksek performanslı REST API'ler inşa etmek için FastAPI ve Pydantic'in nasıl kullanılacağını inceleyeceğiz ki bu API'ler gerçek zamanlı veri uygulamalarını verimli bir şekilde sunabilsin.
FastAPI ve Pydantic Neden?
FastAPI, modern API'ler inşa etmek için en güçlü Python çerçevelerinden biri olarak ortaya çıkmıştır; eşzamanlı programlamanın hızını otomatik API dokümantasyon üretimini birleştirir. Pydantic, veri doğrulama ve serileştirme yetenekleriyle API yanıtınızın hem güvenilir hem de öngörülebilir olmasını sağlar.
Çekirdek Mimarilik
FastAPI uygulamasının çekirdek mimarisini inceleyerek başlayalım. Aşağıdaki örnek, gerçek zamanlı veri API'si için temel bir yapıyı göstermektedir:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List
import asyncio
from datetime import datetime
app = FastAPI(title="Gerçek Zamanlı Veri API'si", 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():
# Gerçek zamanlı veri işleme simülasyonu
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()}
)Performansı Optimize Etme
FastAPI'nin temel güçlerinden biri performans optimizasyon özellikleridir. Gerçek zamanlı veri için eşzamanlı işleme nasıl uygulanır onu inceleyelim:
from fastapi import BackgroundTasks
import asyncio
from concurrent.futures import ThreadPoolExecutor
async def process_real_time_data(data: List[DataPoint]) -> List[DataPoint]:
"""Arka planda ağır işleme simülasyonu"""
# Gerçek senaryolarda bu veritabanı işlemleri, makine öğrenimi çıkarımı veya harici API çağrıları olabilir
await asyncio.sleep(0.1) # Eşzamanlı iş simülasyonu
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
):
# Veriyi eşzamanlı olarak işle
processed_data = await process_real_time_data(data)
# Daha fazla işleme için arka planda görev ekle
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]):
"""Arka planda işleme görevlerini yönet"""
# Veritabanı güncellemeleri, bildirimler veya diğer yan etkiler
passVeri Bütünlüğü İçin Pydantic Doğrulama
Pydantic şema doğrulaması, gerçek zamanlı verilerinizin tutarlılığını ve bütünlüğünü sağlar. Özel doğrulama ile gelişmiş bir örnek:
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('Değer pozitif olmalıdır')
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('Geçersiz kaynak')
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('Sıcaklık sensörleri sıcaklık kategorisine sahip olmalıdır')
return values
class EventStream(BaseModel):
events: List[AdvancedDataPoint]
stream_id: str
last_updated: datetime
class Config:
json_encoders = {
datetime: lambda v: v.isoformat()
}Gerçek Zamanlı WebSocket Entegrasyonu
Gerçek zamanlı uygulamalar için FastAPI ile WebSocket desteğini entegre edin:
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()
# Gerçek zamanlı veri işle
response = {"message": f"Yankı: {data}", "client": client_id}
await manager.send_personal_message(json.dumps(response), websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)İzleme ve Sağlık Kontrolleri
Gerçek zamanlı API'niz için kapsamlı izleme uygulayın:
from fastapi.middleware.tracking import TrackingMiddleware
# İzleme ara katmanı ekle
app.add_middleware(TrackingMiddleware)
@app.get("/health")
async def health_check():
return {
"status": "sağlıklı",
"timestamp": datetime.now().isoformat(),
"services": {
"database": "bağlı",
"cache": "bağlı"
}
}
@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)
}Üretim Dağıtımına Dair Düşünceler
Üretim dağıtımları için Gunicorn ile uvicorn çalışanları kullanmayı düşünün:
# 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 = FalseSonuç
FastAPI ve Pydantic ile gerçek zamanlı veri uygulamaları için yüksek performanslı REST API'ler inşa etmek, geliştiricilere güçlü bir özellik kombinasyonu sunar. Otomatik OpenAPI dokümantasyonu, sağlam veri doğrulama ve eşzamanlı yetenekler bu yığının modern uygulamalar için ideal olduğunu göstermektedir. Bu kılavuzda gösterilen desenleri takip ederek, sadece iyi performanslı değil, aynı zamanda veri bütünlüğünü koruyan ve kapsamlı API dokümantasyonu ile mükemmel geliştirici deneyimi sunan API'ler inşa edebilirsiniz.
IoT veri platformları, finansal ticaret sistemleri veya gerçek zamanlı analiz panoları oluşturuyorsanız, FastAPI ve Pydantic, gerçek zamanlı veri yüklerini verimli bir şekilde işleyebilecek ölçeklenebilir, sürdürülebilir ve yüksek performanslı API'ler inşa etmek için gerekli temeli sunar.