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