文章目錄
- GET方法 StreamingResponse
- POST方法 StreamingResponse
- 其他
- 關于壓縮
GET方法 StreamingResponse
服務器：
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from starlette.responses import FileResponse
from pydantic import BaseModelapp = FastAPI()# 創建一個 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):message: strstatus: 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 uvicornuvicorn.run(app, host="0.0.0.0", port=8000)客戶端：
import requestsurl = "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 iofrom fastapi import FastAPI, File, UploadFile
from fastapi.responses import StreamingResponse
from pydantic import BaseModelapp = FastAPI()# 創建一個 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):message: strstatus: str@app.post("/get_demo_image_with_json")
async def get_demo_image_with_json():# 讀取face.pngimage_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 uvicornuvicorn.run(app, host="0.0.0.0", port=8000)客戶端參數：
import requestsurl = "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 BaseModelapp = FastAPI()# 使用 Gzip 中間件
app.add_middleware(GZipMiddleware, minimum_size=1000, compress_level=6)# 創建一個 Base 模型，用于表示其他的 JSON 信息
class AdditionalInfo(BaseModel):message: strstatus: 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 參數。
)
