FastAPI 教學：回傳 bytes 與 Streaming

簡介

在 Web API 開發中，除了傳回 JSON 之外，二進位資料（如圖片、PDF、音訊）也是常見需求。若直接把完整檔案載入記憶體再回傳，會造成記憶體浪費，甚至在大檔案時導致服務崩潰。FastAPI 提供了 Response、FileResponse、StreamingResponse 等工具，讓開發者能有效且安全地傳送 bytes 或串流給用戶端。

本單元將說明 如何在 FastAPI 中回傳二進位資料、如何使用串流 (Streaming) 來處理大型或即時產生的內容，並提供實作範例、常見陷阱與最佳實踐，幫助你在實務專案中快速上手。

核心概念

1. 直接回傳 `bytes`

最簡單的情況是把二進位資料直接放入 Response 物件的 content 屬性。FastAPI 會自動設定 media_type（MIME type），如果你需要自行指定，只要在 Response 建構子裡傳入 media_type 即可。

範例 1：回傳 PNG 圖片（從檔案讀取）

from fastapi import FastAPI, Response
import pathlib

app = FastAPI()
BASE_DIR = pathlib.Path(__file__).parent

@app.get("/image")
async def get_image():
    # 讀取檔案為 bytes
    img_path = BASE_DIR / "static" / "logo.png"
    img_bytes = img_path.read_bytes()
    # 設定正確的 MIME type
    return Response(content=img_bytes, media_type="image/png")

重點：read_bytes() 會一次把整個檔案載入記憶體，適合小檔案（< 5 MB）或測試用。

2. 使用 `FileResponse` 回傳檔案

若檔案較大，或希望讓瀏覽器直接顯示/下載，FileResponse 是更好的選擇。它會使用 aiofiles（非同步）或 os.sendfile（底層系統呼叫）來效能化傳輸，且支援 range 請求，讓用戶端可以斷點續傳。

範例 2：回傳 PDF 並支援下載

from fastapi import FastAPI
from fastapi.responses import FileResponse
import pathlib

app = FastAPI()
BASE_DIR = pathlib.Path(__file__).parent

@app.get("/report")
async def download_report():
    pdf_path = BASE_DIR / "files" / "annual_report.pdf"
    # attachment_filename 會在 Content-Disposition 加上檔名，trigger download
    return FileResponse(
        path=pdf_path,
        media_type="application/pdf",
        filename="AnnualReport_2024.pdf"
    )

小技巧：若要支援 斷點續傳，只要讓瀏覽器發送 Range 標頭，FileResponse 會自動處理，無需額外程式碼。

3. `StreamingResponse`：逐塊傳送資料

當資料產生過程需要 即時計算（例如壓縮、加密、即時轉碼）或檔案過於龐大（> 100 MB）時，使用 生成器 (generator) 或 非同步迭代器 讓 FastAPI 逐塊 (chunk) 推送 給客戶端，降低記憶體佔用。

範例 3：串流 CSV 產生器（即時產生）

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import csv
import io

app = FastAPI()

def csv_generator():
    # 假設從資料庫逐筆讀取
    header = ["id", "name", "score"]
    rows = [(1, "Alice", 85), (2, "Bob", 92), (3, "Charlie", 78)]

    # 建立文字緩衝區，寫入 CSV 格式
    buffer = io.StringIO()
    writer = csv.writer(buffer)
    writer.writerow(header)
    for row in rows:
        writer.writerow(row)
        # 把緩衝區內容取出，清空再繼續寫入
        buffer.seek(0)
        data = buffer.read()
        buffer.truncate(0)
        yield data.encode()          # 必須回傳 bytes
    # 最後一次確保資料被送出
    buffer.close()

@app.get("/export")
async def export_csv():
    return StreamingResponse(
        csv_generator(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=report.csv"}
    )

說明：csv_generator 每次產生一行 CSV，並以 yield 回傳 bytes。FastAPI 會自動把每個 yield 的結果寫入回應流，客戶端會即時收到資料。

4. 非同步串流：大檔案分段讀取

如果檔案非常龐大，甚至超過磁碟快取，同步 read() 可能會阻塞事件迴圈。此時可結合 aiofiles 以非同步方式讀取檔案，再用 StreamingResponse 串流回傳。

範例 4：非同步串流大型影片檔

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import aiofiles

app = FastAPI()
CHUNK_SIZE = 1024 * 1024  # 1 MB

async def async_file_iterator(file_path: str):
    async with aiofiles.open(file_path, "rb") as f:
        while chunk := await f.read(CHUNK_SIZE):
            yield chunk

@app.get("/video")
async def stream_video():
    video_path = "media/big_movie.mp4"
    return StreamingResponse(
        async_file_iterator(video_path),
        media_type="video/mp4",
        headers={"Accept-Ranges": "bytes"}  # 讓瀏覽器支援快轉/暫停
    )

重點：使用 :=（海象運算子）只在 Python 3.8+ 可用，CHUNK_SIZE 可根據需求調整。Accept-Ranges 讓前端播放器知道伺服器支援斷點請求。

5. 結合 BackgroundTask 處理後續工作

有時候在回傳檔案後，需要執行清理或統計工作（例如刪除臨時檔、寫入下載紀錄）。FastAPI 的 BackgroundTasks 允許你在回應已送出後，非同步執行這些任務，避免阻塞主流程。

範例 5：下載完畢後刪除臨時壓縮檔

from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import FileResponse
import pathlib, os

app = FastAPI()
BASE_DIR = pathlib.Path(__file__).parent

def remove_file(path: str):
    try:
        os.remove(path)
    except OSError:
        pass

@app.get("/download-zip")
async def download_zip(background_tasks: BackgroundTasks):
    zip_path = BASE_DIR / "tmp" / "data_bundle.zip"
    background_tasks.add_task(remove_file, str(zip_path))
    return FileResponse(
        path=zip_path,
        media_type="application/zip",
        filename="data_bundle.zip"
    )

說明：background_tasks.add_task 會把 remove_file 加入事件迴圈的工作列，等回應送出後才執行，確保檔案在傳輸完成前不會被刪除。

常見陷阱與最佳實踐

陷阱	可能的後果	解決方案
一次性讀取大型檔案 (`read_bytes()`、`read()`)	記憶體瞬間飆升，導致 OOM (Out‑Of‑Memory)	改用 `FileResponse` 或 `StreamingResponse`，搭配分塊讀取
忘記設定 `media_type`	瀏覽器無法正確顯示或下載檔案	明確指定 MIME，如 `image/png`、`application/pdf`、`video/mp4`
未處理 `Range` 請求	大檔案在前端播放時無法快轉、暫停	使用 `FileResponse`（自動支援）或自行在 `StreamingResponse` 加上 `Accept-Ranges`
同步迭代器阻塞事件迴圈	其他請求被延遲，效能下降	使用 `aiofiles` 或非同步生成器 (`async def`)
回傳 `str` 而非 `bytes`	文字會被自動編碼為 UTF‑8，可能破壞二進位結構	確保 `yield` 或 `content` 為 `bytes`
未釋放檔案資源	檔案被鎖定、磁碟空間浪費	使用 `with`/`async with` 讓檔案自動關閉

最佳實踐

小檔案 (< 5 MB)：直接 Response(content=bytes, media_type=…) 最簡潔。
中等檔案 (5 MB ~ 100 MB)：使用 FileResponse；若需自訂檔名或額外 Header，直接傳入 filename、headers。
大型檔案 (> 100 MB) 或即時產生：採用 StreamingResponse，配合分塊、非同步 讀取。
需要事後清理：加入 BackgroundTasks，避免在回傳前就刪除檔案。
安全性：永遠檢查路徑是否在允許的目錄內，避免路徑穿越攻擊 (../)。

實際應用場景

場景	為什麼需要回傳 bytes / Streaming	推薦的 FastAPI 方法
圖片縮圖服務	前端需要即時取得縮圖，檔案大小約 30 KB	`Response` + `media_type="image/jpeg"`
報表匯出 (CSV / Excel)	報表內容根據使用者條件即時產生，可能包含上千筆資料	`StreamingResponse` 搭配生成器
影片播放平台	影片檔案數 GB，需要支援快轉、暫停	`FileResponse` 或 `StreamingResponse` + `Accept-Ranges`
大型資料備份下載	使用者點擊一次產生臨時壓縮檔，下載完畢後即刪除	`FileResponse` + `BackgroundTasks`
即時音訊轉碼	音訊來源是外部 API，需在伺服器上即時轉成 MP3 並串流給前端	非同步 `StreamingResponse` + `aiofiles` 或外部轉碼程式

總結

在 FastAPI 中 回傳二進位資料 並不只是把檔案讀進記憶體這麼簡單。透過 Response、FileResponse、StreamingResponse，我們可以依據檔案大小、產生方式與效能需求，選擇最合適的傳輸策略：

小檔案 → Response 直接回傳 bytes。
中等檔案 → FileResponse 自動支援斷點續傳與快取。
大型或即時產生 → StreamingResponse 搭配同步或非同步生成器，降低記憶體佔用。
後續清理 → 使用 BackgroundTasks，確保資源在傳輸完成後正確釋放。

掌握這些概念後，你就能在 API 設計、檔案服務、媒體串流 等場景中，寫出既高效又安全的 FastAPI 端點。祝開發順利，持續探索 FastAPI 更多強大功能吧！ 🚀