FastAPI（56）- 使用 Websocket 打造一个迷你聊天室
2024-04-15 03:49:45
背景
- 在实际项目中，可能会通过前端框架使用 WebSocket 和后端进行通信
- 这里就来详细讲解下 FastAPI 是如何操作 WebSocket 的
模拟 WebSocket 客户端
#!usr/bin/env python
# -*- coding:utf-8 _*-
"""
# author: 小菠萝测试笔记
# blog: https://www.cnblogs.com/poloyy/
# time: 2021/10/5 5:26 下午
# file: 46_websocket.py
"""
import uvicorn
from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponseapp = FastAPI()html = """
<!DOCTYPE html>
<html><head><title>小菠萝聊天室</title></head><body><h1>小菠萝聊天室</h1><form action="" onsubmit="sendMessage(event)"><input type="text" id="messageText" autocomplete="off"/><button>Send</button></form><ul id='messages'></ul><script>// 加载页面，自动创建一个 WebSocket 连接var ws = new WebSocket("ws://localhost:8080/ws");// 收到消息ws.onmessage = function(event) {// 获取输入框的值var messages = document.getElementById('messages')// 创建一个 li 元素var message = document.createElement('li')// 接收 event 的 datavar 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>
"""# 返回一段 HTML 代码给前端
@app.get("/")
async def get():return HTMLResponse(html)@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):# 1、ws 连接await websocket.accept()while True:# 2、接收客户端发送的内容data = await websocket.receive_text()# 3、服务端发送内容await websocket.send_text(f"小菠萝收到的消息是: {data}")if __name__ == '__main__':uvicorn.run(app="46_websocket:app", reload=True, host="127.0.0.1", port=8080)
启动 uvicorn 服务器，访问 127.0.0.1:8080/
客户端、服务端建立 WebSocket 连接成功
发送聊天信息
每发一条消息，均会显示在列表中
可以在其他地方使用 WebSocket
- Depends
- Security
- Cookie
- Header
- Path
- Query
在依赖项中使用 WebSocket
from typing import Optional
import uvicorn
from fastapi import FastAPI, WebSocket, Cookie, Query, status, Dependsfrom fastapi.responses import HTMLResponseapp = FastAPI()html = """
<!DOCTYPE html>
<html><head><title>Chat</title></head><body><h1>小菠萝聊天室</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:8080/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: Optional[str] = Cookie(None),token: Optional[str] = Query(None)
):# 模拟：如果 session 和 token 都为空，则关闭 websocketif session or token:return session or tokenawait websocket.close(code=status.WS_1008_POLICY_VIOLATION)@app.websocket("/items/{item_id}/ws")
async def websocket_depends(websocket: WebSocket,item_id: str,q: Optional[str] = None,# 依赖项cookie_or_token: str = Depends(get_cookie_or_token)
):# 1、创建 websocket 连接await websocket.accept()while True:# 2、接收客户端发送的内容data = await websocket.receive_text()# 3、服务端发送内容await websocket.send_text(f"cookie or token value is:{cookie_or_token}")if q:# 4、如果有传查询参数 q，则再发一条await websocket.send_text(f"query param value is:{q}")# 5、最后再发一条信息await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")if __name__ == '__main__':uvicorn.run(app="46_websocket:app", reload=True, host="127.0.0.1", port=8080)
发送聊天信息
不带查询参数 q
带查询参数 q
当 WebSocket 连接关闭时
await websocket.receive_text() 将引发 WebSocketDisconnect 异常，这不是期望看到的结果
处理断开连接和多个客户端
from typing import Listimport uvicorn
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, statusfrom fastapi.responses import HTMLResponseapp = 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:8080/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>
"""# 返回一段 HTML 代码给前端
@app.get("/")
async def get():return HTMLResponse(html)# 处理和广播消息到多个 WebSocket 连接
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.websocket("/ws/{client_id}")
async def websocket_endpoint(client_id: str, websocket: WebSocket):# 1、客户端、服务端建立 ws 连接await manager.connect(websocket)# 2、广播某个客户端进入聊天室await manager.broadcast(f"{client_id} 进入了聊天室")try:while True:# 3、服务端接收客户端发送的内容data = await websocket.receive_text()# 4、广播某个客户端发送的消息await manager.broadcast(f"{client_id} 发送消息：{data}")# 5、服务端回复客户端await manager.send_personal_message(f"服务端回复{client_id}：你发送的信息是：{data}", websocket)except WebSocketDisconnect:# 6、若有客户端断开连接，广播某个客户端离开了manager.disconnect(websocket)await manager.broadcast(f"{client_id} 离开了聊天室")if __name__ == '__main__':uvicorn.run(app="48_websocket_handler:app", reload=True, host="127.0.0.1", port=8080)
- 模拟一个小型聊天室的场景
- 新的客户端进来，所有人都会收到新客户端进入聊天室的消息
- 某个客户端发送消息，所有人都能看到
- 某个客户端退出了(关闭浏览器)，所有人都会收到该客户端退出聊天室的消息
浏览器打开三个 tab 均访问 127.0.0.1:8080
关掉其中一个客户端(tab)
FastAPI（56）- 使用 Websocket 打造一个迷你聊天室相关推荐
- 教你从零开始用WebSocket打造一个IM聊天室
之前我们在 IM即时聊天室(一):WebSocket 和 IM即时聊天室(二): Socket.io + Node.js 两篇文章中介绍了搭建一个IM的所需的技术栈和通信原理.那在这篇文章里我们就来详 ...
- Vue全家桶+Socket.io+Koa2打造一个智能聊天室 接口已开放
Vue.js+Socket.io+Koa2打造一个智能聊天室 Vue.js全家桶+Socket.io+Express/Koa2 打造的一个智能聊天室. 已经开源啦!为了方便大家学习,智能机器人.IP定 ...
- 如何打造一个语音聊天室
语音聊天室这个名词可能有点陌生,实际上相关的产品还是很多的,例如游戏里的开黑语音.在线课堂等.语音聊天室可以认为视频直播的前身,很多音视频平台的架构是从语音聊天室演进为视频直播室的.本文主要介绍语音聊 ...
- 基于STM32L432KC，通过RT-Thread Studio打造一个迷你时钟
本文中作者基于rt-thread操作系统搭建了一个迷你时钟,用来显示时间和温湿度.rt-thread studio是一款用于rt-thread开发的软件.目前了解到的作用大概为端口设置,代码调试,引入 ...
- vue php聊天室,Laravel + Swoole 打造IM简易聊天室
Laravel + Swoole 打造IM简易聊天室 最近在学习Swoole,利用Swoole扩展让PHP生动了不少,本篇就来Swoole开发一款简易的IM聊天室 应用场景:实现简单的即时消息聊天室. ...
- 连夜撸了一个简易聊天室
点击上方蓝色"方志朋",选择"设为星标" 回复"666"获取独家整理的学习资料! 分不清轮询.长轮询?不知道什么时候该用websocket还 ...
- rudesocket如何使用_[WebSocket入门]手把手搭建WebSocket多人在线聊天室（SpringBoot+WebS...
前言 本文中搭建了一个简易的多人聊天室,使用了WebSocket的基础特性. 源代码来自老外的一篇好文: 本文内容摘要: 初步理解WebSocket的前后端交互逻辑 手把手使用 SpringBoot ...
- 撸一个简易聊天室，不信你学不会实时消息推送（附源码）
点击上方 好好学java ,选择 星标 公众号重磅资讯,干货,第一时间送达 今日推荐:推荐 19 个 github 超牛逼项目!个人原创100W +访问量博客:点击前往,查看更多 分不清轮询.长轮询? ...
- SpringBoot入门建站全系列（二十七）WebSocket做简单的聊天室
SpringBoot入门建站全系列(二十七)WebSocket做简单的聊天室 一.概述 WebSocket 是一种网络通信协议.RFC6455 定义了它的通信标准. WebSocket 是 HTML5 ...
- 实战｜使用 Python 开发一个在线聊天室
「来源: |Python爬虫与数据挖掘 ID:crawler_python」 回复"书籍"即可获赠Python从入门到进阶共10本电子书 今 日 鸡 汤 云想衣裳花想容,春风拂槛露 ...
最新文章
- BPMF论文辅助笔记：采样Ui 部分推导
- system文件_解压MIUI 10 升级包 system.new.dat.br
- ubuntu下vim语法高亮问题
- 分布式事务的特征、原理、以及常见3种解决方案
- 重构是什么、为什么要重构
- git revert reset
- 大数据分析有哪些分析模型
- 学习python3（一）
- java持久化框架对比_四种流行的持久化框架比较
- 精准医学：循环肿瘤DNA在检测非小细胞肺癌患者体细胞突变及跟踪肿瘤进展中的作用｜精准治疗
- 用百度地图开发android程序时，location.getAddrStr()总是为null的解决方法
- PMBOK 项目管理 九大知识领域和五大流程
- python第一天----爬取优美图库的图片
- java entries_Enumerationlt;? extends ZipEntrygt; entries()_学习Java Zip|WIKI教程
- 什么是千兆交换机和万兆交换机？它们有什么区别？
- vue 评论区回复无限嵌套实现代码
- 斯坦福cs224n-2021 assignment1-探索词向量—词共现矩阵—SVD（奇异值分解）
- HTML代码转成字符串数组
- Uva 101：木块问题 The Blocks Problem（详细说明）+（难点）
- 2019/09/03最新进展