Module 12 Exercises: Building Real-Time Systems

Real-time communication is the ultimate test of an API's durability. These exercises will help you bridge the gap between static pages and living, breathing applications.

---

Exercise 1: The Echo Chamber

Create a simple FastAPI WebSocket endpoint at /echo.

• It should accept the connection.
• It should listen for any text message the client sends.
• It should send the exact same text back with the prefix "Echo: ".
• Handle the disconnection properly.

---

Exercise 2: Implementing a Room Guard

Modify the ConnectionManager pattern to include a password for specific rooms.

1. The client sends a room_password as a query parameter in the handshake (e.g., /ws/room101?pass=secret).
2. The server checks the password before calling websocket.accept().
3. If the password is wrong, close the connection with a 4003 (Forbidden) code.

---

Exercise 3: The Dashboard Pulse

You are building a Server Monitoring Dashboard. You want to send the current "CPU Usage" to every connected client once every 5 seconds.

Write the code for the while True loop inside your WebSocket endpoint that:

1. Calculates CPU usage (you can just use a random number for now).
2. Sends it to the client.
3. Waits for 5 seconds using asyncio.sleep.

---

Self-Correction / Discussion

Exercise 1 Answer:

@app.websocket("/echo")
async def echo_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        print("Client disconnected")

Exercise 2 Answer:

@app.websocket("/ws/{room_id}")
async def secret_room(websocket: WebSocket, room_id: str, pass: str):
    if room_id == "VIP" and pass != "secret123":
        await websocket.close(code=4003)
        return
    await websocket.accept()
    # ... logic ...

Exercise 3 Answer:

import asyncio
import random

@app.websocket("/stats")
async def stats_ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        cpu = random.randint(1, 100)
        await websocket.send_json({"cpu": cpu})
        await asyncio.sleep(5)

---

Summary of Module 12

You have mastered the most interactive part of the web.

• Persistence: You know how to keep a line open.
• Bidirectional: You can receive and push data simultaneously.
• Security: You know how to guard your WebSocket handshakes.

In Module 13: OpenAPI and Documentation, we will return to the "Surface" of our API, learning how to customize the beautiful interactive docs that FastAPI produces automatically.