Python基于FastAPI和WebSocket實(shí)現(xiàn)實(shí)時(shí)聊天應(yīng)用
更新時(shí)間：2025年04月30日 10:14:32 作者：老大白菜
這篇文章主要為大家詳細(xì)介紹了Python如何基于FastAPI和WebSocket實(shí)現(xiàn)實(shí)時(shí)聊天應(yīng)用,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下
項(xiàng)目簡(jiǎn)介
這是一個(gè)基于 FastAPI 和 WebSocket 實(shí)現(xiàn)的實(shí)時(shí)聊天應(yīng)用，支持一對(duì)一聊天、離線消息存儲(chǔ)等功能。
技術(shù)棧
后端：FastAPI (Python)
前端：HTML、JavaScript、CSS
通信：WebSocket
認(rèn)證：簡(jiǎn)單的 token 認(rèn)證
項(xiàng)目結(jié)構(gòu)
├── main.py # 后端主程序
└── templates/ # 前端模板目錄
└── chat.html # 聊天頁面
詳細(xì)代碼實(shí)現(xiàn)
1. 后端實(shí)現(xiàn) (main.py)
from fastapi import FastAPI, WebSocket, Depends, WebSocketDisconnect
from typing import Dict
import json
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
app = FastAPI()
app.mount("/templates", StaticFiles(directory="templates"), name="templates")
class ConnectionManager:
def __init__(self):
self.active_connections: Dict[str, WebSocket] = {}
self.offline_messages: Dict[str, list] = {} # 離線消息存儲(chǔ)
async def connect(self, user: str, websocket: WebSocket):
await websocket.accept()
self.active_connections[user] = websocket
async def disconnect(self, user: str, websocket: WebSocket):
if user in self.active_connections:
del self.active_connections[user]
async def send_personal_message(self, message: str, to_user: str):
if to_user in self.active_connections:
await self.active_connections[to_user].send_text(message)
return {"success": True}
else:
# 存儲(chǔ)離線消息
if to_user not in self.offline_messages:
self.offline_messages[to_user] = []
self.offline_messages[to_user].append(message)
return {"success": False, "reason": "offline", "stored": True}
async def get_offline_messages(self, user: str):
if user in self.offline_messages:
messages = self.offline_messages[user]
del self.offline_messages[user] # 取出后刪除
return messages
return []
# Token 認(rèn)證
async def get_cookie_or_token(token: str = None):
if not token:
from fastapi import HTTPException
raise HTTPException(status_code=401, detail="未授權(quán)")
return token
# 初始化連接管理器
ws_manager = ConnectionManager()
@app.websocket("/ws/{user}")
async def websocket_one_to_one(
websocket: WebSocket,
user: str,
cookie_or_token: str = Depends(get_cookie_or_token)
):
await ws_manager.connect(user, websocket)
# 發(fā)送離線消息
offline_msgs = await ws_manager.get_offline_messages(user)
for msg in offline_msgs:
await websocket.send_text(msg)
try:
while True:
data = await websocket.receive_text()
message_data = json.loads(data)
# 防止自己給自己發(fā)消息
if message_data["to_user"] == user:
error_msg = {
"type": "error",
"content": "不能發(fā)送消息給自己"
}
await websocket.send_text(json.dumps(error_msg))
continue
response_message = {
"from": user,
"content": message_data["content"],
"timestamp": message_data.get("timestamp"),
"type": "message"
}
# 發(fā)送消息
result = await ws_manager.send_personal_message(
message=json.dumps(response_message),
to_user=message_data["to_user"]
)
# 處理離線消息
if not result["success"]:
if result.get("stored"):
success_msg = {
"type": "info",
"content": f"消息已保存，將在用戶 {message_data['to_user']} 上線時(shí)送達(dá)"
}
await websocket.send_text(json.dumps(success_msg))
except WebSocketDisconnect:
await ws_manager.disconnect(user, websocket)
except Exception as e:
print(f"WebSocket 錯(cuò)誤: {e}")
await ws_manager.disconnect(user, websocket)
@app.get("/", response_class=FileResponse)
async def root():
return "templates/chat.html"
2. 前端實(shí)現(xiàn) (templates/chat.html)
<!DOCTYPE html>
<html>
<head>
<title>WebSocket Chat</title>
<style>
body {
font-family: Arial, sans-serif;
max-width: 800px;
margin: 0 auto;
padding: 20px;
}
#loginSection {
text-align: center;
margin-top: 100px;
}
#chatSection {
display: none;
}
#messages {
list-style-type: none;
padding: 0;
height: 400px;
overflow-y: auto;
border: 1px solid #ccc;
margin-bottom: 20px;
padding: 10px;
}
.message {
margin: 10px 0;
padding: 10px;
border-radius: 5px;
background-color: #f0f0f0;
}
.input-group {
margin-bottom: 10px;
}
input[type="text"] {
padding: 8px;
margin-right: 10px;
width: 200px;
}
button {
padding: 8px 15px;
background-color: #4CAF50;
color: white;
border: none;
border-radius: 4px;
cursor: pointer;
}
button:hover {
background-color: #45a049;
}
#logoutBtn {
background-color: #f44336;
}
#logoutBtn:hover {
background-color: #da190b;
}
.error {
color: red;
margin: 10px 0;
}
.info {
color: blue;
margin: 10px 0;
}
</style>
</head>
<body>
<!-- 登錄部分 -->
<div id="loginSection">
<h1>WebSocket 聊天</h1>
<div class="input-group">
<input type="text" id="loginUsername" placeholder="輸入用戶名" autocomplete="off"/>
<button onclick="login()">登錄</button>
</div>
<div id="loginError" class="error"></div>
</div>
<!-- 聊天部分 -->
<div id="chatSection">
<h1>WebSocket 聊天</h1>
<div id="currentUser"></div>
<form action="" onsubmit="sendMessage(event)">
<div class="input-group">
<input type="text" id="messageText" placeholder="輸入消息" autocomplete="off"/>
<input type="text" id="username" placeholder="接收者用戶名" autocomplete="off"/>
<button type="submit">發(fā)送</button>
</div>
</form>
<button id="logoutBtn" onclick="logout()">退出</button>
<ul id='messages'></ul>
</div>
<script>
let ws = null;
function checkLogin() {
const token = localStorage.getItem('token');
if (token) {
document.getElementById('loginSection').style.display = 'none';
document.getElementById('chatSection').style.display = 'block';
document.getElementById('currentUser').textContent = `當(dāng)前用戶: ${token}`;
connectWebSocket();
} else {
document.getElementById('loginSection').style.display = 'block';
document.getElementById('chatSection').style.display = 'none';
}
}
function login() {
const username = document.getElementById('loginUsername').value.trim();
if (!username) {
document.getElementById('loginError').textContent = '請(qǐng)輸入用戶名';
return;
}
localStorage.setItem('token', username);
checkLogin();
}
function logout() {
if (ws && ws.readyState === WebSocket.OPEN) {
ws.close();
}
localStorage.removeItem('token');
document.getElementById('messages').innerHTML = '';
document.getElementById('messageText').value = '';
document.getElementById('username').value = '';
document.getElementById('loginSection').style.display = 'block';
document.getElementById('chatSection').style.display = 'none';
document.getElementById('loginUsername').value = '';
document.getElementById('currentUser').textContent = '';
}
function connectWebSocket() {
const token = localStorage.getItem('token');
ws = new WebSocket(`ws://localhost:8000/ws/${token}?token=${token}`);
ws.onmessage = function(event) {
const messages = document.getElementById('messages');
const message = document.createElement('li');
message.className = 'message';
const data = JSON.parse(event.data);
if (data.type === 'error') {
message.style.color = 'red';
message.textContent = data.content;
} else if (data.type === 'info') {
message.style.color = 'blue';
message.textContent = data.content;
} else {
message.textContent = `${data.from}: ${data.content}`;
}
messages.appendChild(message);
messages.scrollTop = messages.scrollHeight;
};
ws.onerror = function(error) {
console.error('WebSocket 錯(cuò)誤:', error);
};
ws.onclose = function() {
console.log('WebSocket 連接已關(guān)閉');
};
}
function sendMessage(event) {
event.preventDefault();
const messageInput = document.getElementById("messageText");
const usernameInput = document.getElementById("username");
if (!messageInput.value.trim() || !usernameInput.value.trim()) {
return;
}
const messageData = {
content: messageInput.value,
to_user: usernameInput.value,
timestamp: new Date().toISOString()
};
ws.send(JSON.stringify(messageData));
const messages = document.getElementById('messages');
const message = document.createElement('li');
message.className = 'message';
message.style.backgroundColor = '#e3f2fd';
message.textContent = `我: ${messageInput.value}`;
messages.appendChild(message);
messages.scrollTop = messages.scrollHeight;
messageInput.value = '';
}
checkLogin();
</script>
</body>
</html>
功能特點(diǎn)
1.用戶管理
- 簡(jiǎn)單的用戶名登錄系統(tǒng)
- 用戶在線狀態(tài)管理
- 用戶會(huì)話保持
2.消息功能
- 實(shí)時(shí)一對(duì)一聊天
- 離線消息存儲(chǔ)
- 上線自動(dòng)接收離線消息
- 防止自己給自己發(fā)消息
3.界面特性
- 響應(yīng)式設(shè)計(jì)
- 消息實(shí)時(shí)顯示
- 錯(cuò)誤信息提示
- 消息狀態(tài)反饋
4.技術(shù)特性
- WebSocket 實(shí)時(shí)通信
- 狀態(tài)管理
- 錯(cuò)誤處理
- 會(huì)話管理
如何運(yùn)行
1.安裝依賴
pip install fastapi uvicorn
2.啟動(dòng)服務(wù)器
uvicorn main:app --reload
3.訪問應(yīng)用
- 打開瀏覽器訪問 http://localhost:8000
- 輸入用戶名登錄
- 開始聊天
使用流程
1.登錄
- 輸入用戶名
- 點(diǎn)擊登錄按鈕
2.發(fā)送消息
- 輸入接收者用戶名
- 輸入消息內(nèi)容
- 點(diǎn)擊發(fā)送
3.接收消息
- 實(shí)時(shí)接收其他用戶發(fā)送的消息
- 上線時(shí)接收離線消息
4.退出
- 點(diǎn)擊退出按鈕
- 清除登錄狀態(tài)
效果圖
以上就是Python基于FastAPI和WebSocket實(shí)現(xiàn)實(shí)時(shí)聊天應(yīng)用的詳細(xì)內(nèi)容，更多關(guān)于Python FastAPI WebSocket實(shí)時(shí)聊天的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
您可能感興趣的文章:
- python開發(fā)實(shí)例之python使用Websocket庫開發(fā)簡(jiǎn)單聊天工具實(shí)例詳解(python+Websocket+JS)
- Python?實(shí)現(xiàn)簡(jiǎn)單智能聊天機(jī)器人
- Python 實(shí)現(xiàn) WebSocket 通信的過程詳解
- Python如何使用WebSocket實(shí)現(xiàn)實(shí)時(shí)Web應(yīng)用
- Python 框架 FastAPI詳解
- 利用Python編寫一個(gè)簡(jiǎn)單的聊天機(jī)器人
- Python中實(shí)現(xiàn)WebSocket的示例詳解
- Python使用FastAPI制作一個(gè)視頻流媒體平臺(tái)
相關(guān)文章
python?Copula?實(shí)現(xiàn)繪制散點(diǎn)模型
這篇文章主要介紹了python?Copula實(shí)現(xiàn)繪制散點(diǎn)模型，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，需要的朋友可以參考一下2022-07-07
這篇文章主要介紹了python基于WordCloud制作詞云圖,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11
python3實(shí)現(xiàn)UDP協(xié)議的服務(wù)器和客戶端
這篇文章主要為大家詳細(xì)介紹了python3實(shí)現(xiàn)UDP協(xié)議的服務(wù)器和客戶端，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-06-06
Tensor 和 NumPy 相互轉(zhuǎn)換的實(shí)現(xiàn)
本文主要介紹了Tensor 和 NumPy 相互轉(zhuǎn)換的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-02-02
二叉樹是分層數(shù)據(jù)結(jié)構(gòu)，其中每個(gè)父節(jié)點(diǎn)最多有 2 個(gè)子節(jié)點(diǎn)，在今天的文章中，我們將討論一個(gè)在大量技術(shù)編碼面試中出現(xiàn)的重要主題，對(duì)Python二叉樹遍歷相關(guān)知識(shí)感興趣的朋友一起看看吧2022-09-09
本文主要介紹了Django實(shí)現(xiàn)視頻播放的具體示例，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下<BR>2022-05-05
Windows8下安裝Python的BeautifulSoup
這篇文章主要介紹了Windows8下安裝Python的BeautifulSoup,本文著重講解安裝中出現(xiàn)的錯(cuò)誤和解決方法,需要的朋友可以參考下2015-01-01
Python轉(zhuǎn)json時(shí)出現(xiàn)中文亂碼的問題及解決
這篇文章主要介紹了Python轉(zhuǎn)json時(shí)出現(xiàn)中文亂碼的問題及解決方案，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2023-02-02
在本篇文章里，我們給大家整理了關(guān)于python元組打包和解包過程的知識(shí)點(diǎn)內(nèi)容，有興趣點(diǎn)的朋友們可以跟著學(xué)習(xí)下。2021-08-08
在本篇文章里小編給大家整理的是一篇關(guān)于python爬蟲智能翻頁批量下載文件的實(shí)例詳解內(nèi)容，有興趣的朋友們可以學(xué)習(xí)下。2021-02-02