DEV Community

Cover image for REST vs GraphQL vs WebSockets vs Webhooks: A Real-World Decision Guide (With Code)
Rose Wabere
Rose Wabere

Posted on

REST vs GraphQL vs WebSockets vs Webhooks: A Real-World Decision Guide (With Code)

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
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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()
Enter fullscreen mode Exit fullscreen mode

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"}
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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 }
    }
  }
}
Enter fullscreen mode Exit fullscreen mode

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 → orders recursion
  • 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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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 def everywhere 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)