day #48 fastapi #12 ws

Author: stackjay

workspace/7_framework/fastapi/README.md

@@ -109,6 +109,10 @@ uvicorn main:app --reload
   Learn to build a production-ready authentication system with JWT, including project structure, configuration management, password hashing, and dependency injection.
   _Includes: `pydantic-settings`, `passlib`, `python-jose`, SQLAlchemy, and comprehensive testing._
 
+- [Day 12: Real-Time Chat with WebSockets](day12/README.md)
+  Learn how to build a real-time chat application with WebSockets, manage connections, and broadcast messages.
+  _Includes: `WebSocket`, connection management, client-side integration, and `pytest` for WebSockets._
+
 ---
 
 ## Recommended Project Structure

workspace/7_framework/fastapi/day12/README.md

@@ -0,0 +1,131 @@
+# FastAPI Day 12: Real-Time Chat with WebSockets
+
+Welcome to **Day 12** of the FastAPI tutorial series! Today, we're diving into the world of real-time communication with WebSockets. You'll learn how to build a simple chat application where multiple clients can connect and exchange messages in real-time.
+
+---
+
+## What You'll Learn
+
+-   **WebSocket Endpoints**: Implement a WebSocket endpoint in FastAPI to handle persistent connections.
+-   **Connection Management**: Create a `ConnectionManager` to keep track of active WebSocket connections.
+-   **Broadcasting Messages**: Broadcast messages to all connected clients to enable real-time chat functionality.
+-   **Client-Side Integration**: Build a simple HTML and JavaScript client to interact with your WebSocket server.
+-   **Testing WebSockets**: Write tests for your WebSocket endpoint using `pytest` and FastAPI's `TestClient`.
+
+---
+
+## Key Concepts
+
+For this tutorial, we've built a simple chat server with a corresponding client.
+
+-   `src/main.py`: Contains the main FastAPI application, the `ConnectionManager`, and the WebSocket endpoint.
+-   `client/index.html`: A simple HTML file with JavaScript to connect to the WebSocket and handle messaging.
+-   `tests/test_main.py`: Contains tests for the WebSocket endpoint.
+
+### 1. The Connection Manager
+
+To manage all active WebSocket connections, we use a `ConnectionManager` class. This class is responsible for adding new connections, removing disconnected ones, and broadcasting messages to all clients.
+
+```python-beginner/workspace/7_framework/fastapi/day12/server/src/main.py#L4-L18
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: List[WebSocket] = []
+
+    async def connect(self, websocket: WebSocket):
+        await websocket.accept()
+        self.active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket):
+        self.active_connections.remove(websocket)
+
+    async def broadcast(self, message: str):
+        for connection in self.active_connections:
+            await connection.send_text(message)
+```
+
+### 2. The WebSocket Endpoint
+
+The core of our application is the WebSocket endpoint, defined with the `@app.websocket()` decorator. This endpoint handles incoming connections, listens for messages, and broadcasts them to all other clients.
+
+```python-beginner/workspace/7_framework/fastapi/day12/server/src/main.py#L26-L38
+@app.websocket("/ws/{client_id}")
+async def websocket_endpoint(websocket: WebSocket, client_id: int):
+    """
+    This endpoint handles WebSocket connections for the chat room.
+    """
+    await manager.connect(websocket)
+    await manager.broadcast(f"Client #{client_id} has entered the chat")
+    try:
+        while True:
+            data = await websocket.receive_text()
+            await manager.broadcast(f"Client #{client_id}: {data}")
+    except WebSocketDisconnect:
+        manager.disconnect(websocket)
+        await manager.broadcast(f"Client #{client_id} has left the chat")
+```
+
+### 3. Client-Side Implementation
+
+The `client/index.html` file provides a simple user interface for the chat. It uses JavaScript's `WebSocket` API to connect to the server, send messages, and display received messages.
+
+```python-beginner/workspace/7_framework/fastapi/day12/client/index.html#L48-L69
+<script>
+    var clientId = Date.now();
+    document.querySelector("#ws-id").textContent = clientId;
+    var ws = new WebSocket(`ws://localhost:8000/ws/${clientId}`);
+    ws.onmessage = function (event) {
+        var messages = document.getElementById("messages");
+        var message = document.createElement("li");
+        var content = document.createTextNode(event.data);
+        message.appendChild(content);
+        messages.appendChild(message);
+        messages.scrollTop = messages.scrollHeight;
+    };
+    function sendMessage(event) {
+        var input = document.getElementById("messageText");
+        ws.send(input.value);
+        input.value = "";
+        event.preventDefault();
+    }
+</script>
+```
+
+### 4. Testing WebSockets
+
+Testing WebSockets is straightforward with FastAPI's `TestClient`. We can use `client.websocket_connect()` to simulate a client connection and test the message flow.
+
+```python-beginner/workspace/7_framework/fastapi/day12/server/tests/test_main.py#L14-L32
+def test_websocket_broadcast(client):
+    """
+    Test the WebSocket broadcasting functionality.
+    - It connects two clients to the /ws/{client_id} endpoint.
+    - It checks the initial connection messages in the correct order.
+    - Has one client send a message.
+    - Asserts that both clients receive the broadcasted message.
+    """
+    with client.websocket_connect("/ws/1") as websocket1:
+        # Client 1 connects and receives its own connection message
+        assert websocket1.receive_text() == "Client #1 has entered the chat"
+
+        with client.websocket_connect("/ws/2") as websocket2:
+            # Client 2 connects, and both clients receive the announcement
+            assert websocket1.receive_text() == "Client #2 has entered the chat"
+            assert websocket2.receive_text() == "Client #2 has entered the chat"
+
+            # Test broadcasting a message from client 1
+            websocket1.send_text("Hello from client 1")
+            assert websocket1.receive_text() == "Client #1: Hello from client 1"
+            assert websocket2.receive_text() == "Client #1: Hello from client 1"
+```
+
+---
+
+## Next Steps
+
+-   Navigate to the server directory and install the dependencies: `cd server && pip install -r requirements.txt`.
+-   Run the application with `uvicorn src.main:app --reload`.
+-   Open `client/index.html` in your web browser. You can open multiple tabs to simulate multiple clients.
+-   Send messages and see them appear in all open tabs.
+-   Run the automated tests with `python -m pytest`.
+
+---

workspace/7_framework/fastapi/day12/client/index.html

@@ -0,0 +1,65 @@
+<!doctype html>
+<html>
+    <head>
+        <title>FastAPI WebSocket Chat</title>
+        <style>
+            body {
+                font-family: Arial, sans-serif;
+                margin: 2em;
+            }
+            #messages {
+                list-style-type: none;
+                margin: 0;
+                padding: 0;
+                border: 1px solid #ccc;
+                height: 300px;
+                overflow-y: scroll;
+                margin-bottom: 1em;
+            }
+            #messages li {
+                padding: 8px 12px;
+            }
+            #messages li:nth-child(odd) {
+                background: #f9f9f9;
+            }
+            #form {
+                display: flex;
+            }
+            #messageText {
+                flex-grow: 1;
+                padding: 8px;
+            }
+            button {
+                padding: 8px 15px;
+            }
+        </style>
+    </head>
+    <body>
+        <h1>FastAPI WebSocket Chat</h1>
+        <p>Your ID: <span id="ws-id"></span></p>
+        <ul id="messages"></ul>
+        <form id="form" onsubmit="sendMessage(event)">
+            <input type="text" id="messageText" autocomplete="off" />
+            <button>Send</button>
+        </form>
+        <script>
+            var clientId = Date.now();
+            document.querySelector("#ws-id").textContent = clientId;
+            var ws = new WebSocket(`ws://localhost:8000/ws/${clientId}`);
+            ws.onmessage = function (event) {
+                var messages = document.getElementById("messages");
+                var message = document.createElement("li");
+                var content = document.createTextNode(event.data);
+                message.appendChild(content);
+                messages.appendChild(message);
+                messages.scrollTop = messages.scrollHeight;
+            };
+            function sendMessage(event) {
+                var input = document.getElementById("messageText");
+                ws.send(input.value);
+                input.value = "";
+                event.preventDefault();
+            }
+        </script>
+    </body>
+</html>

workspace/7_framework/fastapi/day12/server/__init__.py

@@ -0,0 +1,3 @@
+[pytest]
+asyncio_default_fixture_loop_scope = function
+pythonpath = . src

workspace/7_framework/fastapi/day12/server/pytest.ini

@@ -0,0 +1,4 @@
+fastapi
+uvicorn[standard]
+pytest
+httpx

workspace/7_framework/fastapi/day12/server/requirements.txt

@@ -0,0 +1,44 @@
+from typing import List
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
+
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: List[WebSocket] = []
+
+    async def connect(self, websocket: WebSocket):
+        await websocket.accept()
+        self.active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket):
+        self.active_connections.remove(websocket)
+
+    async def broadcast(self, message: str):
+        for connection in self.active_connections:
+            await connection.send_text(message)
+
+manager = ConnectionManager()
+
+# Create the FastAPI app instance
+app = FastAPI()
+
+@app.get("/")
+def read_root():
+    """
+    A simple root endpoint to confirm the API is running.
+    """
+    return {"status": "API is running"}
+
+@app.websocket("/ws/{client_id}")
+async def websocket_endpoint(websocket: WebSocket, client_id: int):
+    """
+    This endpoint handles WebSocket connections for the chat room.
+    """
+    await manager.connect(websocket)
+    await manager.broadcast(f"Client #{client_id} has entered the chat")
+    try:
+        while True:
+            data = await websocket.receive_text()
+            await manager.broadcast(f"Client #{client_id}: {data}")
+    except WebSocketDisconnect:
+        manager.disconnect(websocket)
+        await manager.broadcast(f"Client #{client_id} has left the chat")

workspace/7_framework/fastapi/day12/server/src/__init__.py

@@ -0,0 +1,32 @@
+import pytest
+from fastapi.testclient import TestClient
+from src.main import app
+
+@pytest.fixture
+def client():
+    """
+    Pytest fixture to create a TestClient for our app.
+    """
+    return TestClient(app)
+
+def test_websocket_broadcast(client):
+    """
+    Test the WebSocket broadcasting functionality.
+    - It connects two clients to the /ws/{client_id} endpoint.
+    - It checks the initial connection messages in the correct order.
+    - Has one client send a message.
+    - Asserts that both clients receive the broadcasted message.
+    """
+    with client.websocket_connect("/ws/1") as websocket1:
+        # Client 1 connects and receives its own connection message
+        assert websocket1.receive_text() == "Client #1 has entered the chat"
+
+        with client.websocket_connect("/ws/2") as websocket2:
+            # Client 2 connects, and both clients receive the announcement
+            assert websocket1.receive_text() == "Client #2 has entered the chat"
+            assert websocket2.receive_text() == "Client #2 has entered the chat"
+
+            # Test broadcasting a message from client 1
+            websocket1.send_text("Hello from client 1")
+            assert websocket1.receive_text() == "Client #1: Hello from client 1"
+            assert websocket2.receive_text() == "Client #1: Hello from client 1"