Tech Tutorial - February 28 2026 173007

Pro tip: When you first test a WebSocket server, use wscat (npm) or websocat (Rust) to verify the handshake before wiring up your Python client.

Core Concepts You’ll Need

Before we dive into code, let’s clarify three building blocks that will appear repeatedly:

• Event Loop: The heart of asyncio, it schedules coroutines, I/O callbacks, and futures.
• WebSocket Protocol: A lightweight framing layer on top of TCP, defined by RFC 6455.
• Message Broker (Optional): For horizontal scaling you may inject a broker like Redis Pub/Sub, but for this tutorial we’ll keep everything in‑process.

Understanding these concepts will make the code feel intuitive rather than magical.

Asyncio Basics

At its simplest, an async function returns a coroutine object. The event loop runs these coroutines until they hit an await point, then switches to another ready coroutine. This cooperative model avoids thread‑level contention and keeps CPU usage low.

import asyncio

async def hello():
    print("Hello")
    await asyncio.sleep(1)   # non‑blocking pause
    print("World")

asyncio.run(hello())

Notice how asyncio.sleep yields control back to the loop instead of blocking the thread. In a WebSocket server, every client connection is just another coroutine awaiting I/O.

WebSocket Handshake

The handshake is a simple HTTP upgrade request. The client sends a Sec-WebSocket-Key header; the server replies with a Sec-WebSocket-Accept derived from that key. Libraries like websockets abstract this away, but knowing the flow helps when debugging connection failures.

Setting Up the Development Environment

First, create a fresh virtual environment and install the required packages. We’ll use websockets for the server/client pair and uvloop for a faster event loop implementation.

python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install websockets uvloop

After activation, you can verify the installation:

python -c "import websockets, uvloop; print('All set!')"

Building the WebSocket Server

The server will accept connections, broadcast incoming JSON payloads to all peers, and maintain a tiny in‑memory store of the latest state for each sensor. This pattern mirrors many real‑time dashboards where the most recent value is displayed to every viewer.

Server Skeleton

import asyncio
import json
import websockets
import uvloop

# Global state shared across connections
STATE = {}

# Set of active WebSocket connections
CLIENTS = set()

async def register(ws):
    CLIENTS.add(ws)
    # Send current state to the newly connected client
    await ws.send(json.dumps({"type": "snapshot", "data": STATE}))
    print(f"Client {ws.remote_address} connected. Total: {len(CLIENTS)}")

async def unregister(ws):
    CLIENTS.remove(ws)
    print(f"Client {ws.remote_address} disconnected. Total: {len(CLIENTS)}")

async def broadcast(message: str):
    if CLIENTS:  # avoid iteration over empty set
        await asyncio.wait([client.send(message) for client in CLIENTS])

async def handler(ws, path):
    await register(ws)
    try:
        async for raw_msg in ws:
            # Expecting JSON messages from sensors or UI
            msg = json.loads(raw_msg)
            if msg.get("type") == "update":
                sensor_id = msg["sensor"]
                value = msg["value"]
                # Update global state
                STATE[sensor_id] = value
                # Broadcast the update to everyone
                await broadcast(json.dumps({
                    "type": "update",
                    "sensor": sensor_id,
                    "value": value
                }))
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        await unregister(ws)

if __name__ == "__main__":
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    start_server = websockets.serve(handler, "0.0.0.0", 8765)
    asyncio.get_event_loop().run_until_complete(start_server)
    print("🚀 WebSocket server listening on ws://0.0.0.0:8765")
    asyncio.get_event_loop().run_forever()

The code is deliberately straightforward: every incoming message triggers a state update and an immediate broadcast. The snapshot message ensures late‑joining clients instantly see the current view.

Pro tip: In production, replace the in‑process STATE dict with a Redis hash. That way multiple server instances stay in sync without extra code.

Running the Server

Save the script as realtime_server.py and execute:

python realtime_server.py

You should see the “🚀 WebSocket server listening…” line. The server now accepts connections on port 8765. Open a second terminal to launch a client.

Creating a Minimal WebSocket Client

Our client will simulate a sensor that pushes a random temperature reading every second. It also listens for updates from other peers, printing them to the console.

Client Implementation

import asyncio
import json
import random
import websockets
import uvloop

SERVER_URI = "ws://localhost:8765"

async def sensor_emitter(ws, sensor_id):
    while True:
        # Simulate a temperature between 20°C and 30°C
        value = round(random.uniform(20, 30), 2)
        payload = {
            "type": "update",
            "sensor": sensor_id,
            "value": value
        }
        await ws.send(json.dumps(payload))
        await asyncio.sleep(1)  # non‑blocking wait

async def listener(ws):
    async for raw_msg in ws:
        msg = json.loads(raw_msg)
        if msg["type"] == "snapshot":
            print("🔄 Snapshot received:", msg["data"])
        elif msg["type"] == "update":
            print(f"📡 {msg['sensor']} → {msg['value']}°C")

async def main():
    async with websockets.connect(SERVER_URI) as ws:
        sensor_task = asyncio.create_task(sensor_emitter(ws, "sensor_A"))
        listen_task = asyncio.create_task(listener(ws))
        await asyncio.gather(sensor_task, listen_task)

if __name__ == "__main__":
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
    asyncio.run(main())

Run the client with python realtime_client.py. You’ll see a stream of temperature values printed every second. Open a second terminal and start another instance, perhaps with a different sensor_id. Both clients will now see each other’s updates in real time.

Pro tip: Use websocat ws://localhost:8765 to manually type JSON messages and observe the broadcast behavior without writing a Python client.

Scaling Beyond a Single Process

While the in‑memory approach works for demos, production workloads often demand horizontal scaling. The primary challenge is keeping the STATE and client list consistent across multiple server processes.

Introducing Redis Pub/Sub

Redis offers a lightweight publish/subscribe mechanism that can replace the direct broadcast call. Each server subscribes to a channel (e.g., updates) and republishes incoming messages to its local clients.

import aioredis

REDIS_CHANNEL = "realtime_updates"

async def redis_publisher(message: str):
    redis = await aioredis.create_redis_pool("redis://localhost")
    await redis.publish(REDIS_CHANNEL, message)
    redis.close()
    await redis.wait_closed()

async def redis_subscriber():
    redis = await aioredis.create_redis_pool("redis://localhost")
    res = await redis.subscribe(REDIS_CHANNEL)
    ch = res[0]
    async for msg in ch.iter(encoding="utf-8"):
        await broadcast(msg)   # broadcast to local clients
    redis.close()
    await redis.wait_closed()

Replace the original broadcast call inside the handler with await redis_publisher(message). Then, at server startup, schedule redis_subscriber() as a background task.

Deploying with Docker Compose

Below is a minimal docker-compose.yml that spins up the WebSocket server, a Redis instance, and a sample client.

version: "3.9"
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  server:
    build: .
    command: python realtime_server.py
    depends_on:
      - redis
    ports:
      - "8765:8765"

  client:
    build: .
    command: python realtime_client.py
    depends_on:
      - server

Place the Dockerfile in the same directory with the Python scripts:

FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir websockets uvloop aioredis
CMD ["python", "realtime_server.py"]

Run docker compose up --build and watch the logs. You’ll see all three containers start, and the client will receive updates even if you scale the server service to two replicas:

  server:
    scale: 2

Redis guarantees that every replica receives the same broadcast, keeping the user experience consistent across the cluster.

Pro tip: For ultra‑low latency (sub‑10 ms) you can enable Redis’s replica‑of in a single‑node mode and tune tcp‑keepalive on both server and client sockets.

Real‑World Use Cases

Now that the core pipeline is in place, let’s explore three concrete scenarios where this pattern shines.

• Live Stock Ticker: Feed market data from a broker API into the server, broadcast price ticks to traders’ dashboards, and store the latest quote in Redis for quick lookup.
• Collaborative Whiteboard: Each drawing stroke is a small JSON payload ({x, y, color}) sent over WebSocket; the server relays it to all participants, preserving the canvas state in memory.
• IoT Sensor Hub: Edge devices push telemetry (temperature, humidity, GPS) to the hub; downstream analytics services subscribe via Redis streams for batch processing.

In each case, the combination of asyncio (non‑blocking I/O), WebSockets (push semantics), and Redis (state sharing) yields a solution that is both simple to reason about and horizontally scalable.

Testing and Monitoring

Robust systems need observability from day one. Below are three quick strategies you can adopt.

Unit Tests with pytest‑asyncio

import asyncio
import json
import pytest
import websockets

@pytest.mark.asyncio
async def test_broadcast():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as client1, \
               websockets.connect(uri) as client2:
        # client1 sends an update
        await client1.send(json.dumps({
            "type": "update",
            "sensor": "test_sensor",
            "value": 42
        }))
        # client2 should receive the same update
        msg = await client2.recv()
        data = json.loads(msg)
        assert data["sensor"] == "test_sensor"
        assert data["value"] == 42

Run the test while the server is active: pytest -q. The test validates that the broadcast mechanism works for multiple connections.

Health Checks

Expose a simple HTTP endpoint (using aiohttp or fastapi) that returns 200 OK when the event loop is alive and the Redis connection is healthy. Load balancers can then route traffic only to healthy instances.

Metrics with Prometheus

Instrument the server to export counters such as connected_clients, messages_received_total, and broadcast_latency_seconds. The prometheus_client library makes this a one‑liner:

from prometheus_client import Counter, start_http_server

CONNECTED = Counter("ws_connected_clients", "Number of active WebSocket clients")
MESSAGES = Counter("ws_messages_received_total", "Total messages received")
# Start metrics server on port 8000
start_http_server(8000)

Increment the counters at appropriate points in your code (e.g., after a client registers or a message is processed). Grafana dashboards can then visualize real‑time load.

Security Considerations

WebSockets inherit the security model of the underlying HTTP connection. Here are three essential safeguards.

1. Use WSS (TLS) in Production: Generate a certificate (or use Let’s Encrypt) and serve the WebSocket over wss://. The websockets library accepts an ssl context argument.
2. Authenticate Clients: Require a JWT token in the Sec-WebSocket-Protocol header or as a query parameter. Reject connections that fail validation.
3. Rate‑Limit Messages: Prevent a rogue client from flooding the server. Use asyncio.Semaphore or a simple token bucket per connection.

Implementing these measures early prevents costly retrofits later on.

Performance Tweaks You Can Apply Today

Even though uvloop already gives you a 2‑3× speed boost over the default loop, there are additional levers you can pull.

• Batch Broadcasts: Instead of sending one message per sensor update, aggregate updates over a 50 ms window and send a single JSON array. This reduces TCP packet overhead.
• Binary Frames: WebSocket supports binary frames. Encode your payload with msgpack for a 30‑40 % size reduction compared to plain JSON.
• Back‑Pressure Handling