# 基于FastAPI和Tortoise-ORM的WebSocket类的封装
**以下是基于FastAPI和Tortoise-ORM的WebSocket类的封装，用于实现数据的实时传输：**
```python
from fastapi import FastAPI, HTTPException, WebSocket
from fastapi.responses import HTMLResponse
import uuid
from typing import List
from tortoise.contrib.fastapi import register_tortoise
from tortoise import fields
from tortoise.models import Model
class ChatHistory(Model):
id = fields.UUIDField(pk=True)
user = fields.CharField(max_length=20)
message = fields.CharField(max_length=200)
class Meta:
table = "chat_history"
class WebSocketManager:
def __init__(self):
self.active_connections = []
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 broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
app = FastAPI()
manager = WebSocketManager()
@app.get("/")
async def index():
return HTMLResponse("""
WebSocket Chat
WebSocket Chat
""")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
message = f"{websocket.client.host}: {data}"
await manager.broadcast(message)
chat = ChatHistory(user="user", message=message)
await chat.save()
except Exception as e:
print(str(e))
finally:
manager.disconnect(websocket)
register_tortoise(
app,
db_url="sqlite://db.sqlite3",
modules={"models": ["__main__"]},
generate_schemas=True,
add_exception_handlers=True,
)
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", reload=True)
```
在这个例子中，我们实现了一个简单的聊天室，用户可以通过WebSocket与服务器端进行交互。当用户发送消息时，服务器会将该消息广播给所有连接到聊天室的用户，并将消息保存到数据库中（使用Tortoise-ORM）。同时，我们可以通过访问
---
# Agent Instructions: Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.
Perform an HTTP GET request on the current page URL with the `ask` query parameter:
```
GET https://close.gitbook.io/yun-wei-bi-ji/python/fastapi/ji-yu-fastapi-he-tortoiseorm-de-websocket-lei-de-feng-zhuang.md?ask=
```
The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.
Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.