Master advanced FastAPI features including background tasks, WebSockets, testing, and deployment
Advanced FastAPI features unlock the full potential of modern web applications. From background processing to real-time communication, these features enable building enterprise-grade applications that scale.
Advanced Capabilities:
What You'll Build:
A real-time trading platform with background price updates, WebSocket feeds, comprehensive testing, and production deployment configuration.
Background tasks enable non-blocking operations for better user experience:
Background Tasks:
1from fastapi import BackgroundTasks
2import asyncio
3from datetime import datetime
4
5async def send_email(email: str, subject: str, body: str):
6 # Simulate email sending
7 await asyncio.sleep(2)
8 print(f"Email sent to {email}: {subject}")
9
10async def process_image(filename: str):
11 # Simulate image processing
12 await asyncio.sleep(5)
13 print(f"Image processed: {filename}")
14
15@app.post("/send-notification/")
16async def send_notification(
17 email: str,
18 background_tasks: BackgroundTasks
19):
20 background_tasks.add_task(
21 send_email,
22 email,
23 "Welcome!",
24 "Thank you for joining us"
25 )
26 return {"message": "Email will be sent"}Celery Integration:
1from celery import Celery
2
3celery_app = Celery(
4 "tasks",
5 broker="redis://localhost:6379",
6 backend="redis://localhost:6379"
7)
8
9@celery_app.task
10def heavy_computation(data):
11 # Long-running task
12 import time
13 time.sleep(10)
14 return {"result": "processed"}
15
16@app.post("/process/")
17async def process_data(data: dict):
18 task = heavy_computation.delay(data)
19 return {"task_id": task.id}WebSockets enable real-time bidirectional communication:
WebSocket Implementation:
1from fastapi import WebSocket, WebSocketDisconnect
2from typing import List
3import json
4
5class ConnectionManager:
6 def __init__(self):
7 self.active_connections: List[WebSocket] = []
8
9 async def connect(self, websocket: WebSocket):
10 await websocket.accept()
11 self.active_connections.append(websocket)
12
13 def disconnect(self, websocket: WebSocket):
14 self.active_connections.remove(websocket)
15
16 async def send_personal_message(self, message: str, websocket: WebSocket):
17 await websocket.send_text(message)
18
19 async def broadcast(self, message: str):
20 for connection in self.active_connections:
21 await connection.send_text(message)
22
23manager = ConnectionManager()
24
25@app.websocket("/ws/{client_id}")
26async def websocket_endpoint(websocket: WebSocket, client_id: int):
27 await manager.connect(websocket)
28 try:
29 while True:
30 data = await websocket.receive_text()
31 await manager.send_personal_message(f"You wrote: {data}", websocket)
32 await manager.broadcast(f"Client #{client_id} says: {data}")
33 except WebSocketDisconnect:
34 manager.disconnect(websocket)
35 await manager.broadcast(f"Client #{client_id} left the chat")Real-time Price Updates:
1import asyncio
2import random
3
4@app.websocket("/price-feed")
5async def price_feed(websocket: WebSocket):
6 await websocket.accept()
7 try:
8 while True:
9 # Simulate price updates
10 prices = {
11 "AAPL": round(random.uniform(150, 200), 2),
12 "GOOGL": round(random.uniform(2000, 3000), 2),
13 "TSLA": round(random.uniform(200, 400), 2),
14 }
15 await websocket.send_json({
16 "timestamp": datetime.utcnow().isoformat(),
17 "prices": prices
18 })
19 await asyncio.sleep(1)
20 except WebSocketDisconnect:
21 passComprehensive testing ensures application reliability:
Test Setup:
1from fastapi.testclient import TestClient
2import pytest
3
4client = TestClient(app)
5
6def test_read_main():
7 response = client.get("/")
8 assert response.status_code == 200
9 assert response.json() == {"message": "Hello World"}
10
11@pytest.fixture
12def test_user():
13 return {
14 "username": "testuser",
15 "email": "[email protected]",
16 "password": "testpass123"
17 }
18
19def test_create_user(test_user):
20 response = client.post("/users/", json=test_user)
21 assert response.status_code == 201
22 data = response.json()
23 assert data["username"] == test_user["username"]
24 assert "id" in dataAsync Testing:
1import pytest
2from httpx import AsyncClient
3
4@pytest.mark.asyncio
5async def test_async_endpoint():
6 async with AsyncClient(app=app, base_url="http://test") as ac:
7 response = await ac.get("/async-endpoint")
8 assert response.status_code == 200
9
10@pytest.mark.asyncio
11async def test_websocket():
12 with client.websocket_connect("/ws") as websocket:
13 websocket.send_text("Hello")
14 data = websocket.receive_text()
15 assert data == "Message text was: Hello"Build a complete real-time trading platform with WebSocket price feeds, background order processing, comprehensive testing, and production deployment configuration.
1# š REAL-TIME TRADING PLATFORM
2# Advanced FastAPI with WebSockets, background tasks, and testing
3
4from fastapi import FastAPI, WebSocket, WebSocketDisconnect, BackgroundTasks, Depends
5from fastapi.middleware.cors import CORSMiddleware
6from pydantic import BaseModel, Field
7from typing import List, Dict, Optional
8from datetime import datetime, timedelta
9import asyncio
10import json
11import random
12import uuid
13from enum import Enum
14
15# FastAPI Application
16app = FastAPI(
17 title="Trading Platform API",
18 description="Real-time trading platform with WebSocket feeds",
19 version="3.0.0"
20)
21
22app.add_middleware(
23 CORSMiddleware,
24 allow_origins=["*"],
25 allow_credentials=True,
26 allow_methods=["*"],
27 allow_headers=["*"],
28)
29
30# Models
31class OrderType(str, Enum):
32 buy = "buy"
33 sell = "sell"
34
35class OrderStatus(str, Enum):
36 pending = "pending"
37 filled = "filled"
38 cancelled = "cancelled"
39
40class Stock(BaseModel):
41 symbol: str
42 name: str
43 current_price: float
44 change: float
45 change_percent: float
46 volume: int
47
48class Order(BaseModel):
49 id: str
50 user_id: str
51 symbol: str
52 order_type: OrderType
53 quantity: int
54 price: float
55 status: OrderStatus
56 created_at: datetime
57 filled_at: Optional[datetime] = None
58
59class OrderCreate(BaseModel):
60 symbol: str = Field(..., min_length=1, max_length=10)
61 order_type: OrderType
62 quantity: int = Field(..., gt=0, le=10000)
63 price: float = Field(..., gt=0)
64
65# WebSocket Connection Manager
66class ConnectionManager:
67 def __init__(self):
68 self.active_connections: Dict[str, WebSocket] = {}
69 self.price_subscribers: List[WebSocket] = []
70
71 async def connect_user(self, websocket: WebSocket, user_id: str):
72 await websocket.accept()
73 self.active_connections[user_id] = websocket
74
75 async def connect_price_feed(self, websocket: WebSocket):
76 await websocket.accept()
77 self.price_subscribers.append(websocket)
78
79 def disconnect_user(self, user_id: str):
80 if user_id in self.active_connections:
81 del self.active_connections[user_id]
82
83 def disconnect_price_feed(self, websocket: WebSocket):
84 if websocket in self.price_subscribers:
85 self.price_subscribers.remove(websocket)
86
87 async def send_personal_message(self, message: dict, user_id: str):
88 if user_id in self.active_connections:
89 websocket = self.active_connections[user_id]
90 await websocket.send_json(message)
91
92 async def broadcast_prices(self, prices: Dict[str, Stock]):
93 if self.price_subscribers:
94 message = {
95 "type": "price_update",
96 "timestamp": datetime.utcnow().isoformat(),
97 "data": {symbol: stock.dict() for symbol, stock in prices.items()}
98 }
99 for websocket in self.price_subscribers[:]: # Copy list to avoid modification issues
100 try:
101 await websocket.send_json(message)
102 except:
103 self.price_subscribers.remove(websocket)
104
105manager = ConnectionManager()
106
107# Mock Data Store
108class TradingDataStore:
109 def __init__(self):
110 self.stocks = {
111 "AAPL": Stock(symbol="AAPL", name="Apple Inc.", current_price=175.50, change=2.30, change_percent=1.33, volume=1000000),
112 "GOOGL": Stock(symbol="GOOGL", name="Alphabet Inc.", current_price=2750.80, change=-15.20, change_percent=-0.55, volume=500000),
113 "TSLA": Stock(symbol="TSLA", name="Tesla Inc.", current_price=245.60, change=8.90, change_percent=3.76, volume=2000000),
114 "MSFT": Stock(symbol="MSFT", name="Microsoft Corp.", current_price=415.30, change=1.80, change_percent=0.43, volume=800000),
115 "AMZN": Stock(symbol="AMZN", name="Amazon.com Inc.", current_price=145.20, change=-2.10, change_percent=-1.43, volume=1200000)
116 }
117 self.orders: Dict[str, Order] = {}
118 self.user_portfolios: Dict[str, Dict[str, int]] = {}
119
120 def get_stock(self, symbol: str) -> Optional[Stock]:
121 return self.stocks.get(symbol.upper())
122
123 def update_stock_price(self, symbol: str, new_price: float):
124 if symbol in self.stocks:
125 stock = self.stocks[symbol]
126 old_price = stock.current_price
127 stock.current_price = new_price
128 stock.change = new_price - old_price
129 stock.change_percent = (stock.change / old_price) * 100
130 stock.volume += random.randint(1000, 10000)
131
132 def create_order(self, order_create: OrderCreate, user_id: str) -> Order:
133 order = Order(
134 id=str(uuid.uuid4()),
135 user_id=user_id,
136 symbol=order_create.symbol.upper(),
137 order_type=order_create.order_type,
138 quantity=order_create.quantity,
139 price=order_create.price,
140 status=OrderStatus.pending,
141 created_at=datetime.utcnow()
142 )
143 self.orders[order.id] = order
144 return order
145
146 def get_user_orders(self, user_id: str) -> List[Order]:
147 return [order for order in self.orders.values() if order.user_id == user_id]
148
149data_store = TradingDataStore()
150
151# Background Tasks
152async def process_order(order_id: str):
153 """Background task to process order"""
154 await asyncio.sleep(random.uniform(1, 3)) # Simulate processing time
155
156 if order_id in data_store.orders:
157 order = data_store.orders[order_id]
158 stock = data_store.get_stock(order.symbol)
159
160 if stock:
161 # Simulate order execution logic
162 market_price = stock.current_price
163 if abs(order.price - market_price) / market_price < 0.05: # 5% tolerance
164 order.status = OrderStatus.filled
165 order.filled_at = datetime.utcnow()
166
167 # Update user portfolio
168 if order.user_id not in data_store.user_portfolios:
169 data_store.user_portfolios[order.user_id] = {}
170
171 portfolio = data_store.user_portfolios[order.user_id]
172 if order.symbol not in portfolio:
173 portfolio[order.symbol] = 0
174
175 if order.order_type == OrderType.buy:
176 portfolio[order.symbol] += order.quantity
177 else:
178 portfolio[order.symbol] -= order.quantity
179
180 # Notify user
181 await manager.send_personal_message({
182 "type": "order_update",
183 "order": order.dict()
184 }, order.user_id)
185
186async def price_updater():
187 """Background task to update stock prices"""
188 while True:
189 for symbol in data_store.stocks:
190 # Random price movement
191 current_price = data_store.stocks[symbol].current_price
192 change_percent = random.uniform(-0.05, 0.05) # Ā±5% max change
193 new_price = current_price * (1 + change_percent)
194 data_store.update_stock_price(symbol, round(new_price, 2))
195
196 # Broadcast price updates
197 await manager.broadcast_prices(data_store.stocks)
198 await asyncio.sleep(1) # Update every second
199
200# Start price updater
201@app.on_event("startup")
202async def startup_event():
203 asyncio.create_task(price_updater())
204
205# Dependencies
206async def get_current_user(user_id: str = "demo_user"):
207 """Mock user authentication"""
208 return {"id": user_id, "username": f"user_{user_id}"}
209
210# REST API Endpoints
211@app.get("/", tags=["Root"])
212async def root():
213 return {
214 "message": "Trading Platform API",
215 "version": "3.0.0",
216 "features": ["Real-time prices", "Order execution", "WebSocket feeds"]
217 }
218
219@app.get("/stocks/", response_model=List[Stock], tags=["Stocks"])
220async def get_stocks():
221 """Get all available stocks"""
222 return list(data_store.stocks.values())
223
224@app.get("/stocks/{symbol}", response_model=Stock, tags=["Stocks"])
225async def get_stock(symbol: str):
226 """Get specific stock information"""
227 stock = data_store.get_stock(symbol)
228 if not stock:
229 raise HTTPException(status_code=404, detail="Stock not found")
230 return stock
231
232@app.post("/orders/", response_model=Order, tags=["Orders"])
233async def create_order(
234 order: OrderCreate,
235 background_tasks: BackgroundTasks,
236 current_user: dict = Depends(get_current_user)
237):
238 """Create a new order"""
239 # Validate stock exists
240 stock = data_store.get_stock(order.symbol)
241 if not stock:
242 raise HTTPException(status_code=404, detail="Stock not found")
243
244 # Create order
245 new_order = data_store.create_order(order, current_user["id"])
246
247 # Process order in background
248 background_tasks.add_task(process_order, new_order.id)
249
250 return new_order
251
252@app.get("/orders/", response_model=List[Order], tags=["Orders"])
253async def get_orders(current_user: dict = Depends(get_current_user)):
254 """Get user's orders"""
255 return data_store.get_user_orders(current_user["id"])
256
257@app.get("/portfolio/", tags=["Portfolio"])
258async def get_portfolio(current_user: dict = Depends(get_current_user)):
259 """Get user's portfolio"""
260 portfolio = data_store.user_portfolios.get(current_user["id"], {})
261 portfolio_value = 0
262
263 detailed_portfolio = []
264 for symbol, quantity in portfolio.items():
265 stock = data_store.get_stock(symbol)
266 if stock and quantity > 0:
267 value = stock.current_price * quantity
268 portfolio_value += value
269 detailed_portfolio.append({
270 "symbol": symbol,
271 "quantity": quantity,
272 "current_price": stock.current_price,
273 "value": value
274 })
275
276 return {
277 "portfolio": detailed_portfolio,
278 "total_value": portfolio_value
279 }
280
281# WebSocket Endpoints
282@app.websocket("/ws/prices")
283async def websocket_prices(websocket: WebSocket):
284 """WebSocket endpoint for real-time price updates"""
285 await manager.connect_price_feed(websocket)
286 try:
287 while True:
288 # Keep connection alive
289 await websocket.receive_text()
290 except WebSocketDisconnect:
291 manager.disconnect_price_feed(websocket)
292
293@app.websocket("/ws/user/{user_id}")
294async def websocket_user(websocket: WebSocket, user_id: str):
295 """WebSocket endpoint for user-specific updates"""
296 await manager.connect_user(websocket, user_id)
297 try:
298 while True:
299 # Handle incoming messages
300 data = await websocket.receive_text()
301 message = json.loads(data)
302
303 if message.get("type") == "ping":
304 await manager.send_personal_message(
305 {"type": "pong", "timestamp": datetime.utcnow().isoformat()},
306 user_id
307 )
308 except WebSocketDisconnect:
309 manager.disconnect_user(user_id)
310
311# Health Check
312@app.get("/health", tags=["Health"])
313async def health_check():
314 """Health check endpoint"""
315 return {
316 "status": "healthy",
317 "timestamp": datetime.utcnow(),
318 "active_connections": len(manager.active_connections),
319 "price_subscribers": len(manager.price_subscribers)
320 }
321
322# Demonstration
323print("š Real-time Trading Platform")
324print("=" * 50)
325
326print("\nš Advanced Features:")
327features = [
328 "Real-time WebSocket price feeds",
329 "Background order processing",
330 "User-specific WebSocket connections",
331 "Portfolio tracking and valuation",
332 "Order execution simulation",
333 "Price update broadcasting",
334 "Connection management",
335 "Async task processing"
336]
337
338for i, feature in enumerate(features, 1):
339 print(f"{i:2d}. {feature}")
340
341print("\nš WebSocket Endpoints:")
342endpoints = [
343 "/ws/prices - Real-time price updates",
344 "/ws/user/{user_id} - User-specific notifications"
345]
346
347for endpoint in endpoints:
348 print(f" ā¢ {endpoint}")
349
350print("\nā” Background Tasks:")
351tasks = [
352 "Order processing and execution",
353 "Price updates every second",
354 "Portfolio value calculations",
355 "User notifications"
356]
357
358for i, task in enumerate(tasks, 1):
359 print(f"{i}. {task}")
360
361print("\nš¼ Trading Features:")
362print("ā¢ Real-time stock price updates")
363print("ā¢ Order creation and tracking")
364print("ā¢ Portfolio management")
365print("ā¢ WebSocket real-time communication")
366print("ā¢ Background order processing")
367print("ā¢ Market simulation")
368
369print("\nš Sample Stock Data:")
370sample_stocks = [
371 {"symbol": "AAPL", "price": 175.50, "change": "+1.33%"},
372 {"symbol": "GOOGL", "price": 2750.80, "change": "-0.55%"},
373 {"symbol": "TSLA", "price": 245.60, "change": "+3.76%"},
374 {"symbol": "MSFT", "price": 415.30, "change": "+0.43%"},
375 {"symbol": "AMZN", "price": 145.20, "change": "-1.43%"}
376]
377
378for stock in sample_stocks:
379 print(f" {stock['symbol']}: ${stock['price']} ({stock['change']})")
380
381print("\nšÆ Enterprise Ready:")
382print("ā¢ Scalable WebSocket connections")
383print("ā¢ Background task processing")
384print("ā¢ Real-time data broadcasting")
385print("ā¢ Connection management")
386print("ā¢ Error handling and recovery")
387print("ā¢ Production deployment ready")š Advanced FastAPI Features Breakdown:
1. WebSocket Implementation:
ConnectionManager: Centralized WebSocket connection management2. Background Tasks:
BackgroundTasks: Non-blocking task execution3. Advanced Patterns:
4. Real-time Features:
5. Testing Strategies:
This implementation demonstrates enterprise-grade real-time application architecture!
Your progress is saved in this browser, on this device only.