happyz
/
fastapi
mirror of https://github.com/tiangolo/fastapi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Remove code examples for Python 3.8 in `websockets`

This commit is contained in:
Yurii Motov 2025-12-11 20:16:38 +01:00
parent 2ae496dff4
commit cc4b6d805a
8 changed files with 8 additions and 190 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
docs/en/docs/advanced/websockets.md
View File

@ -38,13 +38,13 @@ In production you would have one of the options above.
But it's the simplest way to focus on the server-side of WebSockets and have a working example: But it's the simplest way to focus on the server-side of WebSockets and have a working example:
{* ../../docs_src/websockets/tutorial001.py hl[2,6:38,41:43] *} {* ../../docs_src/websockets/tutorial001_py39.py hl[2,6:38,41:43] *}
## Create a `websocket` { #create-a-websocket } ## Create a `websocket` { #create-a-websocket }
In your **FastAPI** application, create a `websocket`: In your **FastAPI** application, create a `websocket`:
{* ../../docs_src/websockets/tutorial001.py hl[1,46:47] *} {* ../../docs_src/websockets/tutorial001_py39.py hl[1,46:47] *}
/// note | Technical Details /// note | Technical Details
@ -58,7 +58,7 @@ You could also use `from starlette.websockets import WebSocket`.
In your WebSocket route you can `await` for messages and send messages. In your WebSocket route you can `await` for messages and send messages.
{* ../../docs_src/websockets/tutorial001.py hl[48:52] *} {* ../../docs_src/websockets/tutorial001_py39.py hl[48:52] *}
You can receive and send binary, text, and JSON data. You can receive and send binary, text, and JSON data.

0
docs_src/websockets/tutorial001.py → docs_src/websockets/tutorial001_py39.py
View File

93
docs_src/websockets/tutorial002_an.py
View File

@ -1,93 +0,0 @@
from typing import Union
from fastapi import (
Cookie,
Depends,
FastAPI,
Query,
WebSocket,
WebSocketException,
status,
)
from fastapi.responses import HTMLResponse
from typing_extensions import Annotated
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<form action="" onsubmit="sendMessage(event)">
<label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>
<label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>
<button onclick="connect(event)">Connect</button>
<hr>
<label>Message: <input type="text" id="messageText" autocomplete="off"/></label>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var ws = null;
function connect(event) {
var itemId = document.getElementById("itemId")
var token = document.getElementById("token")
ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
event.preventDefault()
}
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
async def get_cookie_or_token(
websocket: WebSocket,
session: Annotated[Union[str, None], Cookie()] = None,
token: Annotated[Union[str, None], Query()] = None,
):
if session is None and token is None:
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
return session or token
@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
*,
websocket: WebSocket,
item_id: str,
q: Union[int, None] = None,
cookie_or_token: Annotated[str, Depends(get_cookie_or_token)],
):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(
f"Session cookie or query token value is: {cookie_or_token}"
)
if q is not None:
await websocket.send_text(f"Query parameter q is: {q}")
await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")

0
docs_src/websockets/tutorial002.py → docs_src/websockets/tutorial002_py39.py
View File

83
docs_src/websockets/tutorial003.py
View File

@ -1,83 +0,0 @@
from typing import List
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
<head>
<title>Chat</title>
</head>
<body>
<h1>WebSocket Chat</h1>
<h2>Your ID: <span id="ws-id"></span></h2>
<form action="" onsubmit="sendMessage(event)">
<input type="text" id="messageText" autocomplete="off"/>
<button>Send</button>
</form>
<ul id='messages'>
</ul>
<script>
var client_id = Date.now()
document.querySelector("#ws-id").textContent = client_id;
var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);
ws.onmessage = function(event) {
var messages = document.getElementById('messages')
var message = document.createElement('li')
var content = document.createTextNode(event.data)
message.appendChild(content)
messages.appendChild(message)
};
function sendMessage(event) {
var input = document.getElementById("messageText")
ws.send(input.value)
input.value = ''
event.preventDefault()
}
</script>
</body>
</html>
"""
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.send_personal_message(f"You wrote: {data}", websocket)
await manager.broadcast(f"Client #{client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"Client #{client_id} left the chat")

2
tests/test_tutorial/test_websockets/test_tutorial001.py
View File

@ -2,7 +2,7 @@ import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect from fastapi.websockets import WebSocketDisconnect
from docs_src.websockets.tutorial001 import app from docs_src.websockets.tutorial001_py39 import app
client = TestClient(app) client = TestClient(app)

7
tests/test_tutorial/test_websockets/test_tutorial002.py
View File

@ -5,16 +5,15 @@ from fastapi import FastAPI
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from fastapi.websockets import WebSocketDisconnect from fastapi.websockets import WebSocketDisconnect
from ...utils import needs_py39, needs_py310 from ...utils import needs_py310
@pytest.fixture( @pytest.fixture(
name="app", name="app",
params=[ params=[
"tutorial002", pytest.param("tutorial002_py39"),
pytest.param("tutorial002_py310", marks=needs_py310), pytest.param("tutorial002_py310", marks=needs_py310),
"tutorial002_an", pytest.param("tutorial002_an_py39"),
pytest.param("tutorial002_an_py39", marks=needs_py39),
pytest.param("tutorial002_an_py310", marks=needs_py310), pytest.param("tutorial002_an_py310", marks=needs_py310),
], ],
) )

7
tests/test_tutorial/test_websockets/test_tutorial003.py
View File

@ -4,14 +4,11 @@ from types import ModuleType
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from ...utils import needs_py39
@pytest.fixture( @pytest.fixture(
name="mod", name="mod",
params=[ params=[
pytest.param("tutorial003"), pytest.param("tutorial003_py39"),
pytest.param("tutorial003_py39", marks=needs_py39),
], ],
) )
def get_mod(request: pytest.FixtureRequest): def get_mod(request: pytest.FixtureRequest):
@ -32,13 +29,11 @@ def get_client(mod: ModuleType):
return client return client
@needs_py39
def test_get(client: TestClient, html: str): def test_get(client: TestClient, html: str):
response = client.get("/") response = client.get("/")
assert response.text == html assert response.text == html
@needs_py39
def test_websocket_handle_disconnection(client: TestClient): def test_websocket_handle_disconnection(client: TestClient):
with client.websocket_connect("/ws/1234") as connection, client.websocket_connect( with client.websocket_connect("/ws/1234") as connection, client.websocket_connect(
"/ws/5678" "/ws/5678"