FastAPI Learning Road (44) WebSockets

2024-07-12

Nostra analysis praevia tota erat secundum petitiones http, quare si telae suppellectili possunt, respondetur sic, inspice quomodo eam efficiamus.


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}")

Re vera, valde simplex est. Interfaciem scribemus pro nuntiis recipiendis in fine posteriori. Cum nuntius venit, nuntium processimus. Mandatum acceptum directe ad finem anteriorem reddidimus.

Inspice effectum:

Hoc modo munus telae telae simpliciter efficiendum est. Ita nuntium specificum habemus et in chat claudimus, quomodo agere cum illo?


# 接收特定的消息后，终止该聊天
@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}")

Codex ante-finis mutatus non est, solum posteriori finis mutatus est.

Re vera, valde simplex est ad communicationem interretialem efficiendam hic.

Utique in petitionibus uti possumus

Dependet
Securitas
Crustulum
Header
Semita
Query

Inspicere possumus sequentem exemplum scripturae simplicis backend


# 使用多种参数，比如获取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}")

Etsi backend simpliciter implevimus, videre possumus fastapi habere bonum subsidium pro websocket

Technology sharing

FastAPI Learning Road (44) WebSockets

Personal profile

mihi contactus notitia