Coverage for docs_src/websockets/tutorial002_an.py: 100%
21 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
13from typing_extensions import Annotated 1abcde
15app = FastAPI() 1abcde
17html = """ 1abcde
18<!DOCTYPE html>
19<html>
20 <head>
21 <title>Chat</title>
22 </head>
23 <body>
24 <h1>WebSocket Chat</h1>
25 <form action="" onsubmit="sendMessage(event)">
26 <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>
27 <label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
28 <button onclick="connect(event)">Connect</button>
29 <hr>
30 <label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
31 <button>Send</button>
32 </form>
33 <ul id='messages'>
34 </ul>
35 <script>
36 var ws = null;
37 function connect(event) {
38 var itemId = document.getElementById("itemId")
39 var token = document.getElementById("token")
40 ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);
41 ws.onmessage = function(event) {
42 var messages = document.getElementById('messages')
43 var message = document.createElement('li')
44 var content = document.createTextNode(event.data)
45 message.appendChild(content)
46 messages.appendChild(message)
47 };
48 event.preventDefault()
49 }
50 function sendMessage(event) {
51 var input = document.getElementById("messageText")
52 ws.send(input.value)
53 input.value = ''
54 event.preventDefault()
55 }
56 </script>
57 </body>
58</html>
59"""
62@app.get("/") 1abcde
63async def get(): 1abcde
64 return HTMLResponse(html) 1abcde
67async def get_cookie_or_token( 1abcde
68 websocket: WebSocket,
69 session: Annotated[Union[str, None], Cookie()] = None,
70 token: Annotated[Union[str, None], Query()] = None,
71):
72 if session is None and token is None: 1abcde
73 raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION) 1abcde
74 return session or token 1abcde
77@app.websocket("/items/{item_id}/ws") 1abcde
78async def websocket_endpoint( 1abcde
79 *,
80 websocket: WebSocket,
81 item_id: str,
82 q: Union[int, None] = None,
83 cookie_or_token: Annotated[str, Depends(get_cookie_or_token)],
84):
85 await websocket.accept() 1abcde
86 while True:
87 data = await websocket.receive_text() 1abcde
88 await websocket.send_text( 1abcde
89 f"Session cookie or query token value is: {cookie_or_token}"
90 )
91 if q is not None: 1abcde
92 await websocket.send_text(f"Query parameter q is: {q}") 1abcde
93 await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}") 1abcde