Coverage for docs_src/websockets/tutorial003.py: 100%
32 statements
« prev ^ index » next coverage.py v7.6.1, created at 2025-05-05 00:03 +0000
1from typing import List 1abcdef
3from fastapi import FastAPI, WebSocket, WebSocketDisconnect 1abcdef
4from fastapi.responses import HTMLResponse 1abcdef
6app = FastAPI() 1abcdef
8html = """ 1abcdef
9<!DOCTYPE html>
10<html>
11 <head>
12 <title>Chat</title>
13 </head>
14 <body>
15 <h1>WebSocket Chat</h1>
16 <h2>Your ID: <span id="ws-id"></span></h2>
17 <form action="" onsubmit="sendMessage(event)">
18 <input type="text" id="messageText" autocomplete="off"/>
19 <button>Send</button>
20 </form>
21 <ul id='messages'>
22 </ul>
23 <script>
24 var client_id = Date.now()
25 document.querySelector("#ws-id").textContent = client_id;
26 var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);
27 ws.onmessage = function(event) {
28 var messages = document.getElementById('messages')
29 var message = document.createElement('li')
30 var content = document.createTextNode(event.data)
31 message.appendChild(content)
32 messages.appendChild(message)
33 };
34 function sendMessage(event) {
35 var input = document.getElementById("messageText")
36 ws.send(input.value)
37 input.value = ''
38 event.preventDefault()
39 }
40 </script>
41 </body>
42</html>
43"""
46class ConnectionManager: 1abcdef
47 def __init__(self): 1abcdef
48 self.active_connections: List[WebSocket] = [] 1abcdef
50 async def connect(self, websocket: WebSocket): 1abcdef
51 await websocket.accept() 1ghijkl
52 self.active_connections.append(websocket) 1ghijkl
54 def disconnect(self, websocket: WebSocket): 1abcdef
55 self.active_connections.remove(websocket) 1ghijkl
57 async def send_personal_message(self, message: str, websocket: WebSocket): 1abcdef
58 await websocket.send_text(message) 1ghijkl
60 async def broadcast(self, message: str): 1abcdef
61 for connection in self.active_connections: 1ghijkl
62 await connection.send_text(message) 1ghijkl
65manager = ConnectionManager() 1abcdef
68@app.get("/") 1abcdef
69async def get(): 1abcdef
70 return HTMLResponse(html) 1mnopqr
73@app.websocket("/ws/{client_id}") 1abcdef
74async def websocket_endpoint(websocket: WebSocket, client_id: int): 1abcdef
75 await manager.connect(websocket) 1ghijkl
76 try: 1ghijkl
77 while True:
78 data = await websocket.receive_text() 1ghijkl
79 await manager.send_personal_message(f"You wrote: {data}", websocket) 1ghijkl
80 await manager.broadcast(f"Client #{client_id} says: {data}") 1ghijkl
81 except WebSocketDisconnect: 1ghijkl
82 manager.disconnect(websocket) 1ghijkl
83 await manager.broadcast(f"Client #{client_id} left the chat") 1ghijkl