You have used all of these. But when someone asks you, maybe in an interview, or in a system design meeting, why you chose WebSockets over polling, or webhooks over a queue, can you answer precisely?
This isn't another definitions post. This article is about knowing which tool to reach for and why, with code you can actually use.
Quick mental model before we start:
Communication patterns → REST | GraphQL | WebSockets | Webhooks
Code execution model → async/await
These live at different layers. Conflating them is the most common source of confusion.
async/await: The Foundation, Not the Feature
Let's kill one myth immediately: async/await is not a communication pattern. It's how your server handles waiting.
Every I/O operation — database queries, HTTP calls, file reads — makes your code wait. async/await ensures that waiting doesn't freeze every other user's request.
# BAD: blocks the event loop - all other requests stall for 40ms
@app.get("/order/{id}")
def get_order(id: int):
order = db.execute("SELECT * FROM orders WHERE id = %s", id) # blocking
return order
# GOOD: yields control during the wait - other requests run while DB responds
@app.get("/order/{id}")
async def get_order(id: int):
order = await db.fetch_one("SELECT * FROM orders WHERE id = $1", id)
return order
When it matters most: High-concurrency services. A delivery platform handling 500 simultaneous drivers checking order status. An NGO dashboard pulling live survey data.
The trap: Calling a synchronous library inside an async function. This blocks the entire event loop.
import requests # synchronous - don't use this inside async functions
import httpx # async-capable - use this instead
# WRONG
async def fetch_rate():
r = requests.get("https://api.exchangerate.host/latest") # blocks event loop
return r.json()
# CORRECT
async def fetch_rate():
async with httpx.AsyncClient() as client:
r = await client.get("https://api.exchangerate.host/latest")
return r.json()
REST: Default Choice for Good Reason
Use REST when:
- The client initiates all interactions
- Data doesn't change faster than the user refreshes
- You're building standard CRUD
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Order(BaseModel):
customer_id: int
items: list[str]
total: float
@app.get("/orders/{order_id}")
async def get_order(order_id: int):
order = await db.fetch_one("SELECT * FROM orders WHERE id = $1", order_id)
return order
@app.post("/orders")
async def create_order(order: Order):
result = await db.execute(
"INSERT INTO orders (customer_id, items, total) VALUES ($1, $2, $3) RETURNING id",
order.customer_id, order.items, order.total
)
return {"order_id": result["id"]}
@app.delete("/orders/{order_id}")
async def cancel_order(order_id: int):
await db.execute("UPDATE orders SET status = 'cancelled' WHERE id = $1", order_id)
return {"status": "cancelled"}
Real-world scenario: A SACCO member portal. Members log in, check their loan balance, submit a loan application. All request/response. No data is changing while they're looking at a page. REST is perfect.
Production considerations:
- Add proper HTTP caching headers (
Cache-Control,ETag) - REST can be highly cacheable - Version your APIs (
/v1/orders) from day one - Return proper status codes (201 for created, 404 for not found, 422 for validation errors - FastAPI handles this automatically)
GraphQL: REST With Client-Controlled Queries
Use GraphQL when:
- Multiple clients (mobile, web, third-party integrations) need different shapes of the same data
- You're constantly over-fetching or under-fetching with REST
- You have deeply nested, relational data
# Using Strawberry (best GraphQL library for FastAPI)
import strawberry
from fastapi import FastAPI
from strawberry.fastapi import GraphQLRouter
@strawberry.type
class Order:
id: int
total: float
status: str
@strawberry.type
class User:
id: int
name: str
email: str
orders: list[Order]
@strawberry.type
class Query:
@strawberry.field
async def user(self, id: int) -> User:
return await get_user_with_orders(id)
schema = strawberry.Schema(query=Query)
graphql_router = GraphQLRouter(schema)
app = FastAPI()
app.include_router(graphql_router, prefix="/graphql")
Now a mobile app can request exactly what it needs:
# Mobile - bandwidth-conscious, needs minimal data
query {
user(id: 1) {
name
orders(last: 3) {
id
status
}
}
}
# Web dashboard - needs full details
query {
user(id: 1) {
name
email
orders {
id
total
status
items { name price }
}
}
}
Real-world scenario: An NGO platform serving both field officers on 2G mobile and HQ analysts on desktops. Field officers need lightweight data. Analysts need full datasets. GraphQL lets one API serve both without maintaining separate endpoints.
Production considerations:
- Implement depth limiting to prevent abusive nested queries
- Add query complexity analysis: prevent
user → orders → user → ordersrecursion - GraphQL doesn't cache well at HTTP layer: use DataLoader for N+1 query prevention
- Don't default to GraphQL for simple services: it adds real overhead
WebSockets: When the Server Needs to Talk First
Use WebSockets when:
- Data changes continuously, and the user needs to see updates immediately
- Polling would generate unacceptable load or latency
- Both client and server need to send messages freely
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import dict
app = FastAPI()
# Connection manager for multiple clients
class ConnectionManager:
def __init__(self):
self.active: dict[str, WebSocket] = {}
async def connect(self, user_id: str, websocket: WebSocket):
await websocket.accept()
self.active[user_id] = websocket
def disconnect(self, user_id: str):
self.active.pop(user_id, None)
async def send_to_user(self, user_id: str, message: dict):
ws = self.active.get(user_id)
if ws:
await ws.send_json(message)
async def broadcast(self, message: dict):
for ws in self.active.values():
await ws.send_json(message)
manager = ConnectionManager()
@app.websocket("/ws/track/{driver_id}")
async def track_driver(websocket: WebSocket, driver_id: str):
await manager.connect(driver_id, websocket)
try:
while True:
data = await websocket.receive_json()
# Driver sent a location update - broadcast to assigned rider
if data["type"] == "location_update":
await manager.send_to_user(
data["rider_id"],
{"type": "driver_location", "lat": data["lat"], "lng": data["lng"]}
)
except WebSocketDisconnect:
manager.disconnect(driver_id)
Real-world scenario: A ride-hailing app showing a driver moving on the map in real time. The driver's app sends GPS coordinates every 3 seconds over a persistent WebSocket. The rider's app receives them without polling. This would require: 1,000 REST requests per ride if you used polling at 3-second intervals.
Production considerations:
- Persistent connections consume server resources — plan for horizontal scaling early
- Use Redis Pub/Sub to share WebSocket state across multiple server instances
- Always handle
WebSocketDisconnect— clients drop off constantly (network, battery, background app) - Heartbeats keep connections alive through load balancers that close idle connections after 60 seconds
# Heartbeat to keep connection alive
@app.websocket("/ws/live")
async def live_feed(websocket: WebSocket):
await websocket.accept()
try:
while True:
try:
# Wait for message with 30s timeout
data = await asyncio.wait_for(websocket.receive_json(), timeout=30.0)
await handle_message(data)
except asyncio.TimeoutError:
# Send ping to keep connection alive
await websocket.send_json({"type": "ping"})
except WebSocketDisconnect:
pass
Webhooks: Event Notification Between Services
Use webhooks when:
- An external system needs to notify your system that something happened
- You don't control the other system's push mechanism
- You want event-driven integration without maintaining a persistent connection
import hmac
import hashlib
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = "your_flutterwave_webhook_secret"
def verify_flutterwave_signature(payload: bytes, signature: str) -> bool:
expected = hmac.new(
WEBHOOK_SECRET.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
@app.post("/webhooks/payment")
async def payment_webhook(request: Request):
# 1. Verify the request is actually from Flutterwave
signature = request.headers.get("verif-hash", "")
body = await request.body()
if not verify_flutterwave_signature(body, signature):
raise HTTPException(status_code=401, detail="Invalid signature")
payload = await request.json()
event_id = payload["data"]["id"]
# 2. Idempotency - Flutterwave WILL retry on 500s
if await db.webhook_event_exists(event_id):
return {"status": "already_processed"}
# 3. Acknowledge immediately, process async
# Don't do heavy work here -- return 200 fast, process in background
await background_tasks.add_task(process_payment_event, payload)
await db.mark_webhook_received(event_id)
return {"status": "received"}
async def process_payment_event(payload: dict):
order_id = payload["data"]["meta"]["order_id"]
await db.execute(
"UPDATE orders SET status = 'paid', paid_at = NOW() WHERE id = $1",
order_id
)
await send_confirmation_email(order_id)
Real-world scenario: Your Kenyan e-commerce platform integrates Mpesa via Daraja API. When a customer pays, Safaricom calls your /webhooks/mpesa endpoint with the transaction details. You mark the order paid and send a confirmation SMS. No polling. No persistent connection.
Production considerations:
- Always return 200 fast; the webhook caller will retry if you're slow or error
- Never trust webhook data without verifying the signature
- Log every webhook payload for debugging; payment disputes will happen
- Queue heavy processing (emails, SMS, inventory updates) with a background worker
Real-World Architecture: All Four Together
Here's how a production fintech app in Kenya uses all of these together:
┌──────────────────────────────────────────────────────────┐
│ Mobile/Web Client │
└───────┬───────────────────┬─────────────────┬────────────┘
│ │ │
REST (CRUD) WebSocket GraphQL
POST /loans /ws/notifications /graphql
GET /balance (real-time alerts) (analytics)
│ │ │
┌───────▼───────────────────▼─────────────────▼────────────-┐
│ FastAPI Backend │
│ (all async/await internally) │
└───────┬───────────────────────────────────────────────────┘
│
Webhook receiver
POST /webhooks/mpesa ← Safaricom calls this
POST /webhooks/credit ← Credit bureau calls this
Each pattern handles exactly what it's good at. The async/await inside FastAPI makes sure none of them block each other.
When to Use What: Decision Framework
| Signal | Use |
|---|---|
| Client requests data on demand | REST |
| Multiple clients need different data shapes | GraphQL |
| User needs to see live updates without refreshing | WebSocket |
| External service needs to notify your backend of events | Webhook |
| Your code waits on database, HTTP, or file I/O | async/await |
Hard rules:
- If polling interval < 5 seconds and data changes frequently → switch to WebSocket
- If you have > 3 different client types with different data needs → consider GraphQL
- If you're integrating a payment provider, shipping tracker, or auth service → expect webhooks
- If you're running FastAPI → use
async defeverywhere there's I/O, no exceptions
Common Mistakes (Those That Hurt in Production)
1. requests inside async def: blocks the entire event loop. Use httpx with await.
2. No idempotency on webhook handler: payment events get retried. Without idempotency checks, you'll charge customers twice.
3. WebSocket without reconnection logic: mobile networks drop. Your client-side WebSocket needs exponential backoff reconnection, or users see frozen data silently.
4. Assuming GraphQL is real-time: GraphQL subscriptions require a separate WebSocket-based setup. Standard queries/mutations are still request/response.
5. No signature verification on webhooks: your endpoint is public. Anyone can POST to it. Always verify HMAC signatures.
6. Keeping heavy processing in the webhook handler: the caller expects a fast response. Queue everything with a task worker (Celery, ARQ) and return 200 immediately.
If you found this helpful, please share it with others who it might help. And if you have questions, kindly drop them in the comments below!
Rose Wabere - Data & Analytics Engineer, Nairobi. Building real-world data systems with Python, FastAPI, and whatever tool the problem actually needs.
Top comments (0)