main.py
from typing import Listfrom fastapi import FastAPI, WebSocket, WebSocketDisconnectapp = FastAPI()class ConnectionManager:def __init__(self):# 存放激活的ws连接对象self.active_connections: List[WebSocket] = []async def connect(self, ws: WebSocket):# 等待连接await ws.accept()# 存储ws连接对象self.active_connections.append(ws)def disconnect(self, ws: WebSocket):# 关闭时 移除ws对象self.active_connections.remove(ws)@staticmethodasync def send_personal_message(message: str, ws: WebSocket):# 发送个人消息await ws.send_text(message)async def broadcast(self, message: str):# 广播消息for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()@app.websocket("/ws/{user}")
async def websocket_endpoint(websocket: WebSocket, user: str):await manager.connect(websocket)await manager.broadcast(f"用户{user}进入聊天室")try:while True:data = await websocket.receive_text()await manager.send_personal_message(f"你说了: {data}", websocket)await manager.broadcast(f"用户:{user} 说: {data}")except WebSocketDisconnect:manager.disconnect(websocket)await manager.broadcast(f"用户-{user}-离开")if __name__ == "__main__":import uvicorn# 官方推荐是用命令后启动 uvicorn main:app --host=127.0.0.1 --port=8010 --reloaduvicorn.run(app='main:app', host="127.0.0.1", port=8010, reload=True, debug=True)chat.html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>聊天1</title>
</head>
<body>
<h1>User{{id}} Chat</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://127.0.0.1:5200/api/chat/ws/user{{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>
以上部分是案例
此处是个人案例 通过http传递用户协议到jinjia2模板进行html渲染
from fastapi import APIRouter,WebSocket,WebSocketDisconnect
from app1.core.websocket_config import manager
from starlette.requests import Request
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templateschat_routing=APIRouter(prefix='/chat')@chat_routing.websocket("/ws/{user}")
async def websocket_endpoint(websocket: WebSocket, user: str):await manager.connect(websocket)await manager.broadcast(f"用户{user}进入聊天室")try:while True:data = await websocket.receive_text()await manager.send_personal_message(f"你说了: {data}", websocket)await manager.broadcast(f"用户:{user} 说: {data}")except WebSocketDisconnect:manager.disconnect(websocket)await manager.broadcast(f"用户-{user}-离开")templates = Jinja2Templates(directory="templates")
@chat_routing.get("/chat/{id}")
async def read_item(request: Request, id: str):return templates.TemplateResponse("chat.html", {"request": request, "id": id})html
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><title>聊天1</title>
</head>
<body>
<h1>User{{id}} Chat</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://127.0.0.1:5200/api/chat/ws/user{{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>
https://www.cnblogs.com/CharmCode/p/13552551.html
