GET方法 StreamingResponse
服务器：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from starlette.responses import FileResponse
from pydantic import BaseModel
app = FastAPI()
# 创建一个 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):
message: str
status: str
@app.get("/get_demo_image_with_json")
async def get_demo_image_with_json():
# 从文件中读取字节流
file_path = "face.png"
file_like = open(file_path, mode="rb")
# 模拟其他的 JSON 信息
json_info = AdditionalInfo(message="Image loaded successfully", status="OK")
# 使用 StreamingResponse 返回字节流和其他的 JSON 信息
return StreamingResponse(file_like, media_type="image/jpeg",
headers={
"Additional-Info": json_info.model_dump_json()})
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
客户端：
import requests
url = "http://127.0.0.1:8000/get_demo_image_with_json"
response = requests.get(url)
# 检查请求是否成功
if response.status_code == 200:
# 获取文件的字节流
image_data = response.content
# 处理其他的 JSON 信息
additional_info = response.headers.get("Additional-Info")
# 在此处添加您的处理逻辑，例如保存字节流到文件，解析 JSON 信息等
# ...
print("Image loaded successfully.")
print(f"Additional Info: {additional_info}")
else:
print(f"Failed to fetch image. Status code: {response.status_code}")
POST方法 StreamingResponse
服务器代码：
import io
from fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
app = FastAPI()
# 创建一个 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):
message: str
status: str
@app.post("/get_demo_image_with_json")
async def get_demo_image_with_json():
# 读取face.png
image_data = open("face.png", "rb").read()
# 模拟其他的 JSON 信息
json_info = AdditionalInfo(message="Image loaded successfully", status="OK")
# 使用 StreamingResponse 返回字节流和其他的 JSON 信息
return StreamingResponse(io.BytesIO(image_data), media_type="image/png",
headers={
"Additional-Info": json_info.model_dump_json()})
if __name__ == '__main__':
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
客户端参数：
import requests
url = "http://127.0.0.1:8000/get_demo_image_with_json"
response = requests.post(url)
# 检查请求是否成功
if response.status_code == 200:
# 获取文件的字节流
image_data = response.content
# 写入文件
with open("demo_image_with_json.png", "wb") as fp:
fp.write(image_data)
# 打印其他参数
print(response.headers["Additional-Info"])
else:
print(f"Failed to fetch image. Status code: {response.status_code}")
其他
还有FileResponse、base64。
FileResponse太肤浅，base64对于大文件来说太大。
在 FastAPI 中，返回文件字节流的主要方式包括使用 StreamingResponse 和 FileResponse。这两者都可以用于返回二进制数据，例如图像文件。
StreamingResponse: 适用于以流式方式发送数据，对于大型文件特别有用，因为它允许在数据生成时就开始发送，而不必等到整个数据集都可用。
from fastapi.responses import StreamingResponse @app.get("/get_demo_image") async def get_demo_image(): image_data = open("face.png", "rb").read() return StreamingResponse(io.BytesIO(image_data), media_type="image/png")FileResponse: 适用于返回文件，可以从文件系统路径中读取文件内容，也可以通过
content参数直接传递文件内容。from fastapi.responses import FileResponse @app.get("/get_demo_image") async def get_demo_image(): image_data = open("face.png", "rb").read() return FileResponse(content=image_data, media_type="image/png")
这两种方法都是有效的，并且具体的选择可能取决于你的应用程序的需求和性能考虑。如果你希望以异步方式发送文件，你可能会更喜欢 StreamingResponse。如果你只是从文件系统中返回文件，FileResponse 是一个更简单的选择。
关于压缩
FastAPI本身没有直接支持响应内容压缩的中间件，但你可以通过使用 Starlette 的中间件来实现这一功能。具体来说，Starlette 提供了 `Middleware` 类，你可以使用它来定义自定义中间件。以下是一个简单的例子，演示如何使用 Gzip 中间件来压缩响应内容：
```python
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from starlette.responses import FileResponse
from starlette.middleware.gzip import GZipMiddleware
from pydantic import BaseModel
app = FastAPI()
# 使用 Gzip 中间件
app.add_middleware(GZipMiddleware, minimum_size=1000, compress_level=6)
# 创建一个 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):
message: str
status: str
@app.get("/get_demo_image_with_json")
async def get_demo_image_with_json():
# 从文件中读取字节流
file_path = "face.png"
file_like = open(file_path, mode="rb")
# 模拟其他的 JSON 信息
json_info = AdditionalInfo(message="Image loaded successfully", status="OK")
# 使用 StreamingResponse 返回字节流和其他的 JSON 信息
return StreamingResponse(file_like, media_type="image/jpeg",
headers={
"Additional-Info": json_info.json()})
在上述代码中，通过添加 GZipMiddleware 到 FastAPI 应用中，你启用了 Gzip 压缩。在 StreamingResponse 中返回的字节流会在传输过程中被压缩。请注意，Gzip 压缩可能会增加 CPU 使用，但通常可以显著减小传输的数据量，提高性能。你可以根据需求调整 minimum_size 和 compress_level 参数。
