Coverage for docs_src/websockets/tutorial002.py: 100%
20 statements
« prev ^ index » next coverage.py v7.6.1, created at 2024-08-08 03:53 +0000
1from typing import Union 1abcde
3from fastapi import ( 1abcde
4 Cookie,
5 Depends,
6 FastAPI,
7 Query,
8 WebSocket,
9 WebSocketException,
10 status,
11)
12from fastapi.responses import HTMLResponse 1abcde
14app = FastAPI() 1abcde
16html = """ 1abcde
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("/") 1abcde
62async def get(): 1abcde
63 return HTMLResponse(html) 1abcde
66async def get_cookie_or_token( 1abcde
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: 1abcde
72 raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION) 1abcde
73 return session or token 1abcde
76@app.websocket("/items/{item_id}/ws") 1abcde
77async def websocket_endpoint( 1abcde
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() 1abcde
84 while True:
85 data = await websocket.receive_text() 1abcde
86 await websocket.send_text( 1abcde
87 f"Session cookie or query token value is: {cookie_or_token}"
88 )
89 if q is not None: 1abcde
90 await websocket.send_text(f"Query parameter q is: {q}") 1abcde
91 await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}") 1abcde