FastAPI流式响应与StreamingResponse完全教程

介绍：什么是流式响应？

在Web开发中，流式响应是一种处理数据的方式，允许服务器逐步发送数据到客户端，而不是一次性发送所有内容。这对于处理大文件、实时数据流或需要节省内存的场景非常有用。FastAPI基于Starlette框架，内置支持流式响应，通过StreamingResponse类实现。

为什么使用流式响应？

• 高效内存使用：不加载整个数据到内存，适合大文件传输。
• 实时性：可以发送实时数据流，如视频流或传感器数据。
• 用户体验：客户端可以更快接收部分数据，减少等待时间。

StreamingResponse类详解

在FastAPI中，StreamingResponse是一个用于创建流式响应的类，位于fastapi.responses模块。它允许你定义一个生成器函数，逐步生成内容并发送给客户端。

基本用法

首先，导入必要的模块：

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

简单示例：生成文本流

以下是一个简单示例，生成一个字符串流：

@app.get("/stream")
async def stream_response():
    # 定义一个异步生成器函数
    async def generate_text():
        for i in range(5):
            yield f"数据块 {i}\n"  # 逐步生成数据
            await asyncio.sleep(1)  # 模拟延迟，可选
    
    return StreamingResponse(generate_text(), media_type="text/plain")

在这个例子中：

• generate_text()是一个异步生成器，逐步产生数据。
• StreamingResponse接收这个生成器和媒体类型（如"text/plain"）。
• 当客户端请求/stream端点时，服务器会每秒发送一行文本，直到生成器结束。

处理大文件流

流式响应非常适合发送大文件，例如视频或日志文件：

import aiofiles

@app.get("/file-stream")
async def file_stream():
    async def generate_file():
        async with aiofiles.open("large_file.txt", mode="rb") as f:
            while chunk := await f.read(1024):  # 每次读取1KB
                yield chunk
    
    return StreamingResponse(generate_file(), media_type="application/octet-stream")

这里使用aiofiles异步读取文件，分块发送，避免内存溢出。

实际用例和最佳实践

• 视频流处理：结合StreamingResponse和视频编解码，构建实时流媒体API。
• 数据推送：如WebSocket的替代，用于推送实时更新。
• 日志流：服务器日志可以流式传输到客户端进行监控。

错误处理和优化

• 错误处理：在生成器中添加异常捕获，确保流不会意外中断。
• 设置头信息：可以通过headers参数自定义响应头，如Content-Disposition用于文件下载。
• 性能优化：调整块大小或使用压缩减少带宽使用。

总结

流式响应是FastAPI中一个强大功能，通过StreamingResponse类，你可以轻松构建高效的数据流应用。记住：流式响应适合大数据场景，但要注意异步编程和资源管理。多加练习，你将能更好地利用这一特性提升API性能。

如需更多帮助，请参考FastAPI官方文档或社区资源。