如何在 FastAPI 中有效处理流响应？

使用 FastAPI 处理流式响应

FastAPI 提供 StreamingResponse 类，用于在 API 调用期间将数据流式传输到客户端。虽然此功能旨在以非阻塞方式流式传输数据，但在使用具有阻塞操作或不当使用的生成器函数时，可能会出现问题。

增强流式数据响应

确保成功流式传输，请考虑以下事项：

• GET 请求： 使用 GET 请求而不是 POST 来获取数据。
• 安全凭证： 避免通过查询字符串发送凭证；请改用 headers 或 cookies。
• 非阻塞操作： 如果生成器函数包含阻塞操作，请将其定义为 def （不是异步 def）。或者，将 async def 与异步阻塞操作一起使用。
• 合适的媒体类型：为响应指定不同的媒体类型（例如 application/json 或 text/event-stream）而不是文本/plain，或通过将 X-Content-Type-Options 标头设置为 nosniff 来禁用 MIME 嗅探。这可确保浏览器不会缓冲响应，从而允许数据实时传输。

示例实现

考虑以下 Python 代码：

# app.py
from fastapi import FastAPI, StreamingResponse
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def fake_data_streamer():
    for i in range(10):
        yield b'some fake data\n\n'
        await asyncio.sleep(0.5)

@app.get('/')
async def main():
    return StreamingResponse(fake_data_streamer(), media_type='text/event-stream')
    # or, use:
    '''
    headers = {'X-Content-Type-Options': 'nosniff'}
    return StreamingResponse(fake_data_streamer(), headers=headers, media_type='text/plain')
    '''

# test.py (using httpx)
import httpx

url = 'http://127.0.0.1:8000/'

with httpx.stream('GET', url) as r:
    for chunk in r.iter_raw():  # or, for line in r.iter_lines():
        print(chunk)

此代码演示了如何从 FastAPI 应用程序中的生成器函数流式传输数据并使用 httpx 库使用它。

以上是如何在 FastAPI 中有效处理流响应？的详细内容。更多信息请关注PHP中文网其他相关文章！