Coverage for docs_src/websockets/tutorial003.py: 100%
32 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 03:53 +0000
1from typing import List 1abcde
3from fastapi import FastAPI, WebSocket, WebSocketDisconnect 1abcde
4from fastapi.responses import HTMLResponse 1abcde
6app = FastAPI() 1abcde
8html = """ 1abcde
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: 1abcde
47 def __init__(self): 1abcde
48 self.active_connections: List[WebSocket] = [] 1abcde
50 async def connect(self, websocket: WebSocket): 1abcde
51 await websocket.accept() 1abcde
52 self.active_connections.append(websocket) 1abcde
54 def disconnect(self, websocket: WebSocket): 1abcde
55 self.active_connections.remove(websocket) 1abcde
57 async def send_personal_message(self, message: str, websocket: WebSocket): 1abcde
58 await websocket.send_text(message) 1abcde
60 async def broadcast(self, message: str): 1abcde
61 for connection in self.active_connections: 1abcde
62 await connection.send_text(message) 1abcde
65manager = ConnectionManager() 1abcde
68@app.get("/") 1abcde
69async def get(): 1abcde
70 return HTMLResponse(html) 1abcde
73@app.websocket("/ws/{client_id}") 1abcde
74async def websocket_endpoint(websocket: WebSocket, client_id: int): 1abcde
75 await manager.connect(websocket) 1abcde
76 try: 1abcde
77 while True:
78 data = await websocket.receive_text() 1abcde
79 await manager.send_personal_message(f"You wrote: {data}", websocket) 1abcde
80 await manager.broadcast(f"Client #{client_id} says: {data}") 1abcde
81 except WebSocketDisconnect: 1abcde
82 manager.disconnect(websocket) 1abcde
83 await manager.broadcast(f"Client #{client_id} left the chat") 1abcde