Modern web applications demand fast, reliable, and scalable APIs to handle real-time data processing. In this comprehensive guide, we'll explore how to leverage FastAPI and Pydantic to build high-performance REST APIs that can efficiently serve real-time data applications.
Why FastAPI and Pydantic?
FastAPI has emerged as one of the most powerful Python frameworks for building modern APIs, combining the speed of asynchronous programming with automatic API documentation generation. Pydantic, its companion library, provides robust data validation and serialization capabilities that make your API responses both reliable and predictable.
Core Architecture
Let's start by examining the core architecture of a FastAPI application. The following example demonstrates a basic structure for a real-time data 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():
# Simulate real-time data processing
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()}
)
Optimizing Performance
One of FastAPI's key strengths is its performance optimization features. Here's how to implement asynchronous processing for real-time data:
from fastapi import BackgroundTasks
import asyncio
from concurrent.futures import ThreadPoolExecutor
async def process_real_time_data(data: List[DataPoint]) -> List[DataPoint]:
"""Simulate heavy processing in background"""
# In real scenarios, this could involve database operations,
# machine learning inference, or external API calls
await asyncio.sleep(0.1) # Simulate async work
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
):
# Process data asynchronously
processed_data = await process_real_time_data(data)
# Add to background task for further processing
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]):
"""Handle background processing tasks"""
# Database updates, notifications, or other side effects
pass
Pydantic Validation for Data Integrity
Pydantic's schema validation ensures your real-time data maintains consistency and integrity. Here's an advanced example with custom validation:
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()
}
Real-Time WebSocket Integration
For true real-time applications, integrate WebSocket support with 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()
# Process real-time data
response = {"message": f"Echo: {data}", "client": client_id}
await manager.send_personal_message(json.dumps(response), websocket)
except WebSocketDisconnect:
manager.disconnect(websocket)
Monitoring and Health Checks
Implement comprehensive monitoring for your real-time API:
from fastapi.middleware.tracking import TrackingMiddleware
# Add tracking middleware
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)
}
Production Deployment Considerations
For production deployments, consider using Gunicorn with uvicorn workers:
# 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
Conclusion
Building high-performance REST APIs for real-time data applications with FastAPI and Pydantic provides developers with a powerful combination of features. The automatic OpenAPI documentation, robust data validation, and asynchronous capabilities make this stack ideal for modern applications. By following the patterns demonstrated in this guide, you'll be able to build APIs that not only perform well but also maintain data integrity and provide excellent developer experience through comprehensive API documentation.
Whether you're building IoT data platforms, financial trading systems, or real-time analytics dashboards, FastAPI and Pydantic offer the foundation needed to create scalable, maintainable, and high-performance APIs that can handle real-time data loads efficiently.