FastAPI Learning Road (44) WebSockets

2024-07-12

Unsere vorherige Analyse basierte ausschließlich auf http-Anfragen. Wenn WebSockets dies also unterstützen können, lautet die Antwort „Ja“. Schauen wir uns an, wie wir es implementieren.


from fastapi import WebSocket, FastAPI
from fastapi.responses import HTMLResponse
 
app = FastAPI()
 
html = """
 
<!DOCTYPE html>
 
<html>
 
    <head>
 
        <title>Chat</title>
 
    </head>
 
    <body>
 
        <h1>WebSocket 聊天</h1>
 
        <form action="" onsubmit="sendMessage(event)">
 
            <input type="text" id="messageText" autocomplete="off"/>
 
            <button>Send</button>
 
        </form>
 
        <ul id='messages'>
 
        </ul>
 
        <script>
 
            var ws = new WebSocket("ws://localhost:8000/ws");
 
            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>
 
"""
 
 
@app.get("/")
async def get_page():
    return HTMLResponse(html)
 
 
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"接收到的消息是：{data}")

Tatsächlich ist es sehr einfach, eine Schnittstelle zum Empfangen von Nachrichten im Backend zu schreiben. Wir haben die empfangene Nachricht direkt an das Frontend zurückgegeben.

Werfen wir einen Blick auf die Wirkung:

Auf diese Weise implementieren wir einfach die Funktion eines Websockets. Wir haben also eine bestimmte Nachricht und schließen den Chat. Wie gehen wir damit um?


# 接收特定的消息后，终止该聊天
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        if data == "bye":
            await websocket.send_text(f"请注意：窗口即将关闭")
            await websocket.send_text("窗口已关闭")
            await websocket.close()
            break
        else:
            await websocket.send_text(f"接收到的消息是：{data}")

Der Front-End-Code wurde nicht geändert, nur das Back-End wurde geändert.

Tatsächlich ist es hier sehr einfach, die WebSocket-Kommunikation zu implementieren.

Natürlich können wir es auch in Anfragen verwenden

Kommt darauf an
Sicherheit
Plätzchen
Header
Weg
Abfrage

Wir können uns das folgende Beispiel eines einfachen Backend-Schreibens ansehen


# 使用多种参数，比如获取cookie等
from typing import Optional
from fastapi import Cookie, Query, status, Depends
 
 
async def get_cookie_or_token(
    websocket: WebSocket,
    session: Optional[str] = Cookie(None),
    token: Optional[str] = Query(None),
):
    if session is None and token is None:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
    return session or token
 
 
@app.websocket("item/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    q: Optional[int] = None,
    cookie_or_token: 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"parameter q is: {q}")
        await websocket.send_text(f"Message text was: {data}, for : {q}")

Obwohl wir das Backend einfach implementiert haben, können wir sehen, dass Fastapi WebSocket gut unterstützt

Technologieaustausch

FastAPI Learning Road (44) WebSockets

Persönliches Profil

meine Kontaktdaten