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