Coverage for docs_src/websockets/tutorial002.py: 100%
20 statements
« prev ^ index » next coverage.py v7.6.1, created at 2025-05-05 00:03 +0000
1from typing import Union 1abcdef
3from fastapi import ( 1abcdef
4 Cookie,
5 Depends,
6 FastAPI,
7 Query,
8 WebSocket,
9 WebSocketException,
10 status,
11)
12from fastapi.responses import HTMLResponse 1abcdef
14app = FastAPI() 1abcdef
16html = """ 1abcdef
17<!DOCTYPE html>
18<html>
19 <head>
20 <title>Chat</title>
21 </head>
22 <body>
23 <h1>WebSocket Chat</h1>
24 <form action="" onsubmit="sendMessage(event)">
25 <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>
26 <label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
27 <button onclick="connect(event)">Connect</button>
28 <hr>
29 <label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
30 <button>Send</button>
31 </form>
32 <ul id='messages'>
33 </ul>
34 <script>
35 var ws = null;
36 function connect(event) {
37 var itemId = document.getElementById("itemId")
38 var token = document.getElementById("token")
39 ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);
40 ws.onmessage = function(event) {
41 var messages = document.getElementById('messages')
42 var message = document.createElement('li')
43 var content = document.createTextNode(event.data)
44 message.appendChild(content)
45 messages.appendChild(message)
46 };
47 event.preventDefault()
48 }
49 function sendMessage(event) {
50 var input = document.getElementById("messageText")
51 ws.send(input.value)
52 input.value = ''
53 event.preventDefault()
54 }
55 </script>
56 </body>
57</html>
58"""
61@app.get("/") 1abcdef
62async def get(): 1abcdef
63 return HTMLResponse(html) 1KLMNOP
66async def get_cookie_or_token( 1abcdef
67 websocket: WebSocket,
68 session: Union[str, None] = Cookie(default=None),
69 token: Union[str, None] = Query(default=None),
70):
71 if session is None and token is None: 1yzmngABophCDqriEFstjGHuvkIJwxl
72 raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION) 1zBDFHJ
73 return session or token 1ymngAophCqriEstjGuvkIwxl
76@app.websocket("/items/{item_id}/ws") 1abcdef
77async def websocket_endpoint( 1abcdef
78 websocket: WebSocket,
79 item_id: str,
80 q: Union[int, None] = None,
81 cookie_or_token: str = Depends(get_cookie_or_token),
82):
83 await websocket.accept() 1mngophqristjuvkwxl
84 while True:
85 data = await websocket.receive_text() 1mngophqristjuvkwxl
86 await websocket.send_text( 1mngophqristjuvkwxl
87 f"Session cookie or query token value is: {cookie_or_token}"
88 )
89 if q is not None: 1mngophqristjuvkwxl
90 await websocket.send_text(f"Query parameter q is: {q}") 1ghijkl
91 await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}") 1mngophqristjuvkwxl