File uploads, pagination, and real-time WebSockets are common features in modern web applications that improve usability, performance, and interactivity.
File uploads allow users to send documents, images, or other data from the client to the server for storage or processing. Pagination helps manage large datasets by delivering data in smaller, manageable chunks through APIs or user interfaces.
Real-time WebSockets enable persistent, two-way communication between the client and server, allowing applications to push updates instantly without repeated requests.
File Uploads in FastAPI
File uploads are a staple in web apps, from profile pictures to document sharing. FastAPI simplifies this with built-in support for multipart forms, async handling, and secure validation, making it ideal for high-traffic services.
Handling Single and Multiple File Uploads
FastAPI uses the UploadFile class from Starlette to process files efficiently without loading everything into memory. This prevents crashes during large uploads and supports streaming for better performance.
Start by defining an endpoint with File(...) or UploadFile:
from fastapi import FastAPI, File, UploadFile, HTTPException
from fastapi.responses import JSONResponse
import shutil
import os
app = FastAPI()
@app.post("/upload-single/")
async def upload_single(file: UploadFile = File(...)):
if not file.content_type.startswith("image/"):
raise HTTPException(status_code=400, detail="Only images allowed")
file_path = f"uploads/{file.filename}"
os.makedirs("uploads", exist_ok=True)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {"filename": file.filename, "size": file.size}For multiple files, use List[UploadFile]
@app.post("/upload-multiple/")
async def upload_multiple(files: list[UploadFile] = File(...)):
results = []
for file in files:
# Validation and save logic here
results.append({"name": file.filename})
return {"uploaded": results}Key practices include
1. Client-side: Use FormData in JavaScript (e.g., fetch with multipart/form-data).
2. Security: Validate MIME types, scan for viruses (integrate ClamAV), and enforce size limits via max_file_size.
3. Storage: Save to local disks for dev, but scale to S3 or Cloudinary in production.
Advanced File Processing and Validation
Beyond basics, process files on-the-fly with libraries like Pillow for images or FFmpeg for videos. FastAPI's async nature shines here.
Consider a resize endpoint
1. Receive the file.
2. Use Pillow to resize: img = Image.open(file.file); img.thumbnail((800, 600)).
3. Save optimized version and return metadata.
This setup powers apps like Instagram clones, where uploads must be fast and reliable.
Pagination for Scalable Data Handling
As datasets grow (think millions of users or logs), dumping everything in one response kills performance. Pagination slices data into manageable pages, with FastAPI offering flexible query params for offset, cursor, or page-based schemes.
Implementing Offset and Limit Pagination
Offset pagination uses skip and limit—simple but fine for moderate datasets.
from fastapi import Query
from typing import Optional
from pydantic import BaseModel
class PaginatedResponse(BaseModel):
items: list
total: int
page: int
size: int
@app.get("/users/")
async def get_users(
skip: int = Query(0, ge=0),
limit: int = Query(10, ge=1, le=100)
):
# Simulate DB query, e.g., SQLAlchemy: .offset(skip).limit(limit)
items = fake_users[skip:skip + limit] # Replace with real DB
return PaginatedResponse(
items=items,
total=len(fake_users),
page=skip // limit + 1,
size=limit
)Pros: Intuitive for users. Cons: Inefficient for deep pages due to full table scans.
Cursor-Based Pagination for Large-Scale Apps
For infinite scrolls (e.g., Twitter feeds), use cursors—opaque tokens encoding the last item's ID.
1. Query with cursor (base64-encoded last_id) and limit.
2. Fetch WHERE id > decoded_cursor ORDER BY id LIMIT limit.
3. Return next_cursor for the last item's encoded ID.
Pagination Strategies
Industry Tip: Combine with SQLAlchemy's paginate() or FastAPI-Pagination extension for zero-boilerplate.
Real-Time Communication with WebSockets
WebSockets enable bidirectional, persistent connections—perfect for live chats, notifications, or collaborative editing. FastAPI's WebSocket support is first-class, leveraging ASGI for low-latency updates.
Setting Up WebSocket Endpoints
Define a WebSocket route and handle connect/disconnect
from fastapi import WebSocket, WebSocketDisconnect
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 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()
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)Building a Real-Time Chat Application
Scale it into a chat room
1. Authenticate on connect (JWT in query params).
2. Store messages in Redis for persistence.
3. Use send_json for structured data (e.g., {user: "Alice", msg: "Hi"}).
Enhancements
1. Rooms: Dict of client_id: room_id for segmented broadcasts.
2. Typing Indicators: Send ephemeral events like {"type": "typing", "user": "Bob"}.
3. Fallback: Graceful downgrade to Server-Sent Events (SSE) for firewalls.
Performance Benchmarks
Per 2025 WebSocket benchmarks, FastAPI excels in async Python ecosystems.
Integrating All Features: A Practical Example
Tie it together in a dashboard API: Paginated file lists with live upload notifications.
1. Endpoint: GET /files/?page=1&limit=20 (paginated).
2. Upload: POST /files/ triggers WebSocket broadcast: "New file uploaded!".
3. Real-time: Watch /ws/files for updates without refresh.
Deployment Checklist
1. Use Uvicorn with --workers 4 for concurrency.
2. Redis for pub/sub in WebSockets.
3. NGINX proxy for WebSocket upgrades (proxy_http_version 1.1;).
This holistic integration mirrors apps like Notion or Slack.
Class Sessions
Sales Campaign
We have a sales campaign on our promoted courses and products. You can purchase 1 products at a discounted price up to 15% discount.