超简单的fastapi链接websocket用例

这篇具有很好参考价值的文章主要介绍了超简单的fastapi链接websocket用例。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

main.py

from typing import List

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = 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)

    @staticmethod
    async 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 --reload
    uvicorn.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 Jinja2Templates

chat_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文章来源地址https://www.toymoban.com/news/detail-617957.html

到了这里,关于超简单的fastapi链接websocket用例的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 【vue+websocket】vue本地链接websocket正常,线上部署websocket 无法加载响应数据【已解决】

    在你的服务器的安全组中开放websocket的端口号。如果有,可以忽略当前步骤。 http 情况下用 ws , https 情况下必须要使用 wss 同时,线上的websocket需要用域名( 如果你配置了域名 ) ✨ 踩坑不易,还希望各位大佬支持一下 textcolor{gray}{踩坑不易,还希望各位大佬支持一下} 踩坑不易

    2024年02月16日
    浏览(45)
  • Vue中WebSocket链接中断、心跳机制防止自动断开

    1、WebSocket链接中断原因 WebSocket断开的原因有很多,最好在WebSocket断开时,将错误打印出来。 二、心跳机制防止自动断开 WebSocket在一段时间内没有进行通讯便会自读断开链接,可以每隔30秒或一分钟向服务器发送一次通讯防止链接终端

    2024年02月16日
    浏览(45)
  • 回调函数的简单用例

    ①用MsgInterface_t定义一个结构体s_Lin_MsgInterface,包含两个回调函数成员: ②确定结构体下的回调函数成员的参数: ③传入实参,确定结构体下的回调函数成员的函数名: ④最终回调函数执行的对应函数如下:

    2024年02月14日
    浏览(34)
  • autox.js链接WebSocket实战,需要的老铁直接拿走

     昨天写脚本的时候需要用到sockcet接口,在网上苦寻一番之后发现这个段代码还是比较ok的,今天拿出来给大家分享一下,需要的直接取!!

    2024年04月16日
    浏览(45)
  • Vue 如何使用WebSocket与服务器建立链接 持续保持通信

    浏览器通过JavaScript向服务器发出建立WebSocket链接的请求,链接建立后,客户端和服务器端就可以通过TCP链接直接交互数据。WebSocket链接后可以通过 send() 方法来向服务器发送数据,并通过 onnessage 事件来接受服务器返回的数据。 WebSocket - Web API 接口参考 | MDN

    2024年01月18日
    浏览(40)
  • python系列28:fastapi部署应用

    FastAPI 是一个用于构建 API 的现代、快速(高性能)的 web 框架,类似flask,Django,webpy 在部署时可能需要用到下面的库: Uvicorn 或者 Hypercorn负责ASGI 服务器。 Starlette 负责 web 部分。 Pydantic 负责数据部分。 都用pip install安装即可 示例代码如下: 如果你的代码里会出现 async /

    2024年01月18日
    浏览(37)
  • python高并发优选之FastAPI

    FastAPI是一个基于Python 3.6+的现代Web框架,它专注于高性能和易用性。FastAPI通过结合多种技术实现了出色的性能,包括异步编程、类型提示和自动文档生成。FastAPI基于Starlette框架,并且使用Pydantic库进行数据验证和转换,从而使RESTful API的开发变得更加容易。 在FastAPI中,GET和

    2024年02月11日
    浏览(42)
  • 【项目实战】基于netty-websocket-spring-boot-starter实现WebSocket服务器长链接处理

    项目中需要建立客户端与服务端之间的长链接,首先就考虑用WebSocket,再来SpringBoot原来整合WebSocket方式并不高效,因此找到了netty-websocket-spring-boot-starter 这款脚手架,它能让我们在SpringBoot中使用Netty来开发WebSocket服务器,并像spring-websocket的注解开发一样简单 2.1.1 @ServerEndpo

    2024年02月12日
    浏览(46)
  • Python - 【FastAPI】框架配置搭建基本使用

    之前在面试的时候面试官有提到过这个框架,但是个人在之前的项目中没有用到过,只是有听过,并没有实际的应用,因此,个人自己研究了一下,这是个新型的框架,使用起来相对挺简单。FastAPI是一个基于Python的现代Web框架,它具有快速构建高性能API的特点。 中文文档:

    2024年02月11日
    浏览(35)
  • websocket链接断开服务器报错java.io.EOFException解决办法

    首先说一下报错原因,由于默认60s无消息交互就会关闭长连接,所以导致这个报错 解决方案其实也很简单,既然是长时间没有通信导致断开,其实保持通信就好,可以设置心跳时间来解决。 可以前端也可以后端,附上前端代码吧 //心跳检测 var heartCheck = {     timeout: 60000,//

    2024年02月08日
    浏览(55)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包