快捷導(dǎo)航

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ì)一聊天、離線(xiàn)消息存儲(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 # 聊天頁(yè)面

詳細(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] = {}  # 離線(xiàn)消息存儲(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ǔ)離線(xiàn)消息
            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ā)送離線(xiàn)消息
    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"]
            )
            
            # 處理離線(xiàn)消息
            if not result["success"]:
                if result.get("stored"):
                    success_msg = {
                        "type": "info",
                        "content": f"消息已保存，將在用戶(hù) {message_data['to_user']} 上線(xiàn)時(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="輸入用戶(hù)名" 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="接收者用戶(hù)名" 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)前用戶(hù): ${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)輸入用戶(hù)名';
                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.用戶(hù)管理

簡(jiǎn)單的用戶(hù)名登錄系統(tǒng)
用戶(hù)在線(xiàn)狀態(tài)管理
用戶(hù)會(huì)話(huà)保持

2.消息功能

實(shí)時(shí)一對(duì)一聊天
離線(xiàn)消息存儲(chǔ)
上線(xiàn)自動(dòng)接收離線(xiàn)消息
防止自己給自己發(fā)消息

3.界面特性

響應(yīng)式設(shè)計(jì)
消息實(shí)時(shí)顯示
錯(cuò)誤信息提示
消息狀態(tài)反饋

4.技術(shù)特性

WebSocket 實(shí)時(shí)通信
狀態(tài)管理
錯(cuò)誤處理
會(huì)話(huà)管理

如何運(yùn)行

1.安裝依賴(lài)

pip install fastapi uvicorn

2.啟動(dòng)服務(wù)器

uvicorn main:app --reload

3.訪問(wèn)應(yīng)用

打開(kāi)瀏覽器訪問(wèn) http://localhost:8000
輸入用戶(hù)名登錄
開(kāi)始聊天

使用流程

1.登錄

輸入用戶(hù)名
點(diǎn)擊登錄按鈕

2.發(fā)送消息

輸入接收者用戶(hù)名
輸入消息內(nèi)容
點(diǎn)擊發(fā)送

3.接收消息

實(shí)時(shí)接收其他用戶(hù)發(fā)送的消息
上線(xiàn)時(shí)接收離線(xiàn)消息

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)文章！

您可能感興趣的文章: