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