23.3 实时文档协作
FastAPI实时文档协作完整教程：掌握WebSockets与Python实现
这是一份详细的FastAPI教程，教你如何使用Python和WebSockets构建实时文档协作应用，适合新手学习，涵盖环境搭建、代码实现和最佳实践。
推荐工具
FastAPI实时文档协作教程：从基础到实战
引言：什么是实时文档协作？
实时文档协作允许多个用户同时在线编辑同一文档，变更立即同步，适用于团队办公和远程协作。本教程将使用FastAPI和WebSockets在Python中实现此功能，让新学习者轻松上手。
1. 环境设置与FastAPI基础
首先，确保安装Python 3.7+和FastAPI。运行以下命令：
pip install fastapi uvicorn websockets
创建一个简单的FastAPI应用：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"message": "欢迎学习FastAPI实时协作"}
启动服务器：uvicorn main:app --reload
2. 集成WebSockets实现实时通信
FastAPI支持WebSockets，这是实时协作的核心。我们将创建一个WebSocket端点来处理用户连接和消息。
from fastapi import WebSocket
from fastapi.responses import HTMLResponse
html = """
<!DOCTYPE html>
<html>
<head>
<title>实时协作</title>
</head>
<body>
<textarea id="editor" rows="20" cols="80"></textarea>
</body>
</html>
"""
@app.get("/")
async def get():
return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"已接收: {data}")
3. 处理多人编辑与数据库
为了支持多用户协作，我们需要一个数据库来存储文档状态。使用SQLite作为简单示例，推荐生产环境中用PostgreSQL或MongoDB。
安装数据库驱动：
pip install sqlalchemy databases
创建模型和连接：
from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.ext.declarative import declarative_base
from databases import Database
DATABASE_URL = "sqlite:///./test.db"
database = Database(DATABASE_URL)
Base = declarative_base()
class Document(Base):
__tablename__ = "documents"
id = Column(Integer, primary_key=True, index=True)
content = Column(Text, default="")
在FastAPI应用启动时初始化数据库。
4. 实现实时协作逻辑
扩展WebSocket处理函数，以广播更新到所有连接的客户端。这需要一个连接管理器来追踪用户。
from fastapi import WebSocket
from typing import List
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
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)
manager = ConnectionManager()
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
# 保存到数据库（省略详细代码）
# 例如，await database.execute("UPDATE documents SET content = :content WHERE id = 1", {"content": data})
await manager.broadcast(data) # 广播给所有用户
except Exception as e:
manager.disconnect(websocket)
5. 冲突解决与前端优化
在多人编辑时，可能发生冲突。简单解决方案是使用最后写入优先或乐观锁。前端可以通过WebSocket接收更新并合并变更。
推荐使用OT（操作转换）算法处理复杂冲突，但为简化，教程使用定时同步或提示用户保存。
6. 安全与部署考虑
- 添加身份验证（如JWT）来控制访问。
- 使用HTTPS确保安全。
- 部署到云服务如Heroku或AWS，并配置负载均衡。
总结
本教程介绍了如何使用FastAPI和WebSockets构建实时文档协作应用，涵盖从环境设置到核心实现。通过实践，你可以扩展功能如版本控制或用户权限。
进一步学习：参考FastAPI官方文档和开源项目，如Firepad。保持代码模块化，确保项目可维护。
开发工具推荐