首页 > 其他分享 >fastapi_socketio_chatroom

fastapi_socketio_chatroom

时间：2023-01-30 09:12:22浏览次数：59  
标签：socket socketio fastapi app chatroom sid document

E:\song2\fastapi_socketio_chatroom\app.py

#  socketio
import socketio

# fastapi
from fastapi import FastAPI, WebSocket, Request, WebSocketDisconnect
from fastapi.responses import RedirectResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi.templating import Jinja2Templates
from fastapi.middleware.cors import CORSMiddleware

# docs_url一定要设置成None,才会使用本地的 swagger-ui 的静态文件
app = FastAPI(docs_url=None)

# socketio
sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins=[])
app.mount("/ws", socketio.ASGIApp(sio))

# cros
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# template and static
templates = Jinja2Templates(directory="templates")
app.mount('/static', StaticFiles(directory='static'), name='static')

# docs
@app.get('/docs', include_in_schema=False)
async def custom_swagger_ui_html():
    return get_swagger_ui_html(
        openapi_url=app.openapi_url,
        title=app.title + " - Swagger UI ",
        oauth2_redirect_url=app.swagger_ui_oauth2_redirect_url,
        swagger_js_url="/static/swagger-ui/swagger-ui-bundle.js",
        swagger_css_url="/static/swagger-ui/swagger-ui.css",
        swagger_favicon_url="/static/swagger-ui/favicon.png"
    )


# docs文档
@app.get("/", tags=["Docs"])
async def get_docs():
    return RedirectResponse("/docs")


# jinjia2的测试网页
@app.get("/agv", response_class=HTMLResponse)
async def read_jinjia2(request: Request):
    return templates.TemplateResponse("index.html", {"request": request })


# websocket的测试网页
@app.get("/websocket", response_class=HTMLResponse)
async def websocket_test(request: Request):
    return templates.TemplateResponse("websocket.html", {"request": request})

# websocket接口
@app.websocket("/wstest")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"send from serve: {data}")

# socketio 事件

# 连接 事件
@sio.event
async def connect(sid, environ):
    # 连接成功之后,会自动恢复一个connect事件给前端
    print(f'~~~~sid={sid}, {environ["HTTP_ORIGIN"]},连接成功~~~~')
    sio.enter_room(sid,'agv')
    await sio.emit('message',f'sid={sid},进入房间',room='agv')
    
# 断开事件
@sio.event
async def disconnect(sid):
    print(f'~~~~sid={sid},断开连接~~~~')
    sio.leave_room(sid,'agv')
    await sio.emit('disconnect',f'sid={sid},离开房间',room='agv')

# 消息事件
@sio.event
async def message(sid,data):
    print(f'~~~~data = {data}, message from sid={sid}~~~~')
    await sio.emit('message','msg from server',room='agv')

E:\song2\fastapi_socketio_chatroom\main.py

import uvicorn

if __name__ == '__main__':
    uvicorn.run('app:app', host="127.0.0.1", port=9000, reload=True)

E:\song2\fastapi_socketio_chatroom\templates\index.html

<!DOCTYPE html>
<html lang="en">

<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>AGV Path</title>
  <script type="module" crossorigin src="{{url_for('static',path='/assets/index.85cdc1d7.js' )}}"></script>
  <link rel="stylesheet" href="{{url_for('static',path='/assets/index.c1b68991.css' )}}">
</head>

<body>
  <div id="app"></div>

</body>

</html>

E:\song2\fastapi_socketio_chatroom\templates\websocket.html

<!DOCTYPE html>
<html>

<head>
    <title>Chat</title>
</head>

<body>
    <h1>WebSocket Chat</h1>
    <form action="" onsubmit="sendMessage(event)">
        <input type="text" id="messageText" autocomplete="off" />
        <button>Send</button>
    </form>
    <ul id='messages'>
    </ul>
    <script>
        // socket的连接地址
        const ws = new WebSocket("ws://localhost:9000/ws");

        // socket 连接成功的回调
        ws.addEventListener('open', (e) => {
            console.log(`socket 连接成功`)
        })
        // socket 连接失败的回调
        ws.addEventListener('error', (e) => {
            console.log(`socket 连接失败`)
        })
        // socket 连接断开时候的回调
        ws.addEventListener('close', (e) => {
            console.log(`socket 连接断开`)
        })

        // 采用这中属性赋值的方法,只能有一个回调函数,如果想要有多个回调函数,可以采用上面的addEventListener的方法,来添加回调函数
        ws.onmessage = function (event) {
            receMessage(event.data)
        };

        // 将接受到内容渲染到界面上
        function receMessage(msg) {
            var messages = document.getElementById('messages')
            var message = document.createElement('li')
            var content = document.createTextNode(msg)
            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>

E:\song\socketio_test\vite-project-2\index.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <link rel="icon" type="image/svg+xml" href="/vite.svg" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>Vite + TS</title>
  </head>
  <body>
    <div id="app"></div>
    <script type="module" src="/src/main.ts"></script>
  </body>
</html>

E:\song\socketio_test\vite-project-2\src\main.ts

import { io } from "socket.io-client";
import axios from "axios";

const axiosInstance = axios.create({
  baseURL: "http://127.0.0.1:9000",
  timeout: 1000,
});

const socket = io("http://localhost:9000/", { path: "/ws/socket.io/" });

document.querySelector<HTMLDivElement>("#app")!.innerHTML = `
  <div>
 	<button id='send'>发送axios请求</button> 
  <hr/>
 	<button id='open-socket'>打开socket连接</button> 
 	<button id='send-socket'>发送socket请求</button> 
 	<button id='close-socket'>断开socket连接</button> 
  </div>
`;

document.getElementById("send")!.addEventListener("click", async () => {
  const res = await axiosInstance.get("/httptest");
  const oRes = document.createElement("div");
  oRes.innerHTML = res.data.data;
  document.querySelector<HTMLDivElement>("#app")!.appendChild(oRes);
});

document.getElementById("open-socket")!.addEventListener("click", async () => {
  socket.connect();
});

document.getElementById("send-socket")!.addEventListener("click", async () => {
  const msg = "msg from client";
  socket.emit("message", msg);
  const oRes = document.createElement("div");
  oRes.innerHTML = `发送: ${msg}`;
  document.querySelector<HTMLDivElement>("#app")!.appendChild(oRes);
});

document.getElementById("close-socket")!.addEventListener("click", async () => {
  socket.close();
});

socket.on("connect", () => {
  console.log("~~connect success~~");
});

socket.on("disconnect", (response: any) => {
  console.log("~~  disconnect ~~" + response);
});

socket.on("message", (response) => {
  console.log(`收到: ${response}`);

  const oRes = document.createElement("div");
  oRes.innerHTML = `收到: ${response}`;
  document.querySelector<HTMLDivElement>("#app")!.appendChild(oRes);
});

标签：socket,socketio,fastapi,app,chatroom,sid,document
From： https://www.cnblogs.com/zhuoss/p/17074297.html

相关文章

  • fastapi socketio cros问题 已解决
    代码importsocketiofromfastapiimportFastAPIfromfastapi.middleware.corsimportCORSMiddlewarefromfastapi.responsesimportRedirectResponse#appapp=......
  • fastapi socketio http cros solved
    D:\code_gitee\fastapi-socketio-example-main\fastapi-socketio-example-main\app.pyimportosimportpathlibimportsecretsimporttimefromtypingimportOption......
  • axios_socketio_cros_test
    D:\code_gitee\html_crostest\htmo_crostest\index.html<!DOCTYPEhtml><htmllang="en"><head><metacharset="UTF-8"/><linkrel="icon"type="image/sv......
  • flask_socketio_cors
    D:\code_gitee\flask_socketio_online-ping-demo-master\online-ping-demo-master\python3\app.py#!/usr/bin/python3#-*-coding:utf-8-*-fromflaskimportcopy_c......
  • fastapi_socketio
    代码逻辑图E:\song\下载的文件\fastapi-socketio-example-master\backend\asgi.pyimportuvicornfromapiimportconfig,appif__name__=='__main__':uvic......
  • Vue3请求Fastapi接口提示No 'Access-Control-Allow-Origin' header is present...
    问题出现在使用Vue3和Fastapi做前后端分离项目时，前端调用接口console报错：No'Access-Control-Allow-Origin'headerispresentontherequestedresource解决方法在......
  • FastAPI学习
      fromfastapiimportFastAPIfromenumimportEnumfrompydanticimportBaseModelfromtypingimportUnionapp=FastAPI()classModelName(str,Enum):......
  • fastapi
    fastapiFastAPI是一个用于构建API的现代、快速（高性能）的web框架，使用Python3.6+并基于标准的Python类型提示。关键特性:快速：可与NodeJS和Go比肩的极高性......
  • FastAPI 记录笔记
    https://fastapi.tiangolo.com/安装pipinstallfastapipipinstall"uvicorn[standard]"基本代码main.pyfromtypingimportUnionfromfastapiimportFastAPI......
  • agv_fastapi_socket_mock
    E:\song\agv_fastapi_socket_v4_mock\app.pyimporttimefromfastapiimportFastAPI,WebSocket,Request,WebSocketDisconnectfromfastapi.responsesimportRedire......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99