FastAPI 如何在FastAPI应用程序中发送操作进度

在本文中，我们将介绍如何在FastAPI应用程序中发送操作进度。FastAPI是一个高性能的Web框架，它使用Python 3.7+中的类型提示来构建API，并具有自动交互式文档功能。通过使用FastAPI，我们可以轻松地创建功能强大的API，并且还可以在操作执行的过程中向客户端发送操作进度。接下来，我们将详细介绍如何实现这一点。

阅读更多：FastAPI 教程

使用WebSocket发送进度

一种常见的方法是使用WebSocket来实现操作进度的发送。FastAPI内置了WebSocket功能，使我们能够轻松地实现这一目标。我们可以通过创建一个WebSocket连接并在操作执行的过程中发送进度消息。

首先，我们需要在FastAPI应用程序中添加WebSocket路由。我们可以使用websocket_manager来创建WebSocket连接并管理连接。

from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect

app = FastAPI()

class ConnectionManager:
    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 send_message(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

connection_manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await connection_manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await connection_manager.send_message(data)
    except WebSocketDisconnect:
        connection_manager.disconnect(websocket)

在上面的代码中，我们创建了一个名为ConnectionManager的类来管理WebSocket连接。它具有connect、disconnect和send_message方法，用于连接、断开连接和发送消息。在websocket_endpoint函数中，我们通过connection_manager来处理WebSocket连接的创建和关闭，并在接收到消息时发送消息给所有活动连接。

接下来，我们可以使用这个WebSocket连接来发送操作进度。假设我们有一个耗时较长的操作，我们希望在操作执行的过程中向客户端发送进度。我们可以在执行操作的循环中使用await connection_manager.send_message(progress_data)来发送进度消息。

@app.get("/long_operation")
async def long_operation():
    for i in range(10):
        # 执行操作的代码

        progress_data = f"Progress: {i * 10}%"
        await connection_manager.send_message(progress_data)

    return {"message": "Operation completed"}

在上面的代码中，我们模拟了一个耗时较长的操作，循环10次并在每次循环中发送进度消息。在实际应用中，您将需要根据操作的性质和持续时间来定义自己的操作进度逻辑。

使用HTTP长轮询发送进度

除了使用WebSocket，我们还可以使用HTTP长轮询（HTTP long polling）来发送操作进度。HTTP长轮询是一种技术，客户端发送一个HTTP请求并在服务器上保持连接打开，直到服务器有新数据可用时返回响应。客户端在收到响应后，可以立即发送下一个请求以保持连接。

要使用HTTP长轮询发送进度，我们可以创建一个API端点，客户端可以发送一个HTTP请求并保持连接打开以等待进度更新。当有进度更新时，服务器将返回最新的进度数据。

import asyncio
from fastapi import FastAPI

app = FastAPI()

progress_data = None
connections = []

@app.get("/progress")
async def get_progress():
    async with asyncio.Lock() as lock:
        connection = asyncio.Event()
        connections.append(connection)

        while progress_data is None:
            await connection.wait()

        response = {"progress": progress_data}
        progress_data = None

        return response

@app.post("/progress")
async def post_progress(progress_value: int):
    global progress_data
    progress_data = progress_value

    for connection in connections:
        connection.set()

    return {"message": "Progress updated"}

在上面的例子中，我们使用asyncio库和asyncio.Lock来处理多个客户端连接的同步问题。当一个客户端发送一个HTTP GET请求时，它会等待直到有进度数据可用时才返回响应。当一个客户端发送一个HTTP POST请求时，它将更新进度数据并通知所有等待的客户端。

总结

本文介绍了如何在FastAPI应用程序中发送操作进度。我们使用了两种方法：使用WebSocket和使用HTTP长轮询。通过这些方法，我们可以轻松地向客户端发送操作执行的进度，以提供更好的用户体验。无论您选择哪种方法，都应该根据您的应用程序需求和客户端要求来选择最合适的方法。无论您使用哪种方法，FastAPI的强大功能和性能将确保操作进度的高效传输。