FastAPI – 表單與檔案上傳（Form & File Upload）

主題：非同步檔案讀寫

---

簡介

在現代的 Web 應用程式中，檔案上傳已成為必不可少的功能。無論是使用者上傳大尺寸的圖片、影片，或是系統需要接收 CSV、JSON 等資料檔案，都必須確保上傳流程既安全又高效。FastAPI 以其原生支援非同步（async）程式設計而聞名，使得在處理大量檔案時不會阻塞伺服器的 I/O，提升整體效能與使用者體驗。

本篇文章聚焦於 非同步檔案讀寫 的實作技巧，從基礎概念、實作範例，到常見陷阱與最佳實踐，帶領讀者一步一步建立可在生產環境直接使用的 FastAPI 檔案上傳 API。

---

核心概念

1. 為什麼要使用非同步 I/O？

• 避免阻塞：傳統的同步 I/O（如 open(...).read()）會在讀寫檔案期間阻塞整個執行緒，若同時有多個請求，會造成效能瓶頸。
• 更佳的併發：使用 async / await 搭配 aiofiles 等非同步檔案庫，讓單一執行緒可以同時處理多個 I/O 操作。
• 相容於 ASGI：FastAPI 基於 ASGI（Asynchronous Server Gateway Interface），非同步函式能直接與伺服器協同工作，達到最大化效能。

2. UploadFile 與 File 物件

FastAPI 提供兩種接收檔案的方式：

類別	說明	何時使用
File	以 bytes 形式載入整個檔案，適合小檔案（< 1 MB）	快速驗證、簡單處理
UploadFile	以 SpooledTemporaryFile 包裝，支援非同步讀寫	大檔案、需要流式處理

重點：若未特別指定 UploadFile，FastAPI 會自動使用 File，因此在需要非同步操作時，務必使用 UploadFile。

3. 非同步讀寫檔案的工具

套件	用途	安裝指令
aiofiles	提供 async 版的 open, read, write 等檔案操作	pip install aiofiles
python-multipart	FastAPI 解析 multipart/form-data 必備	pip install python-multipart

---

程式碼範例

以下示範 4 個常見且實用的非同步檔案讀寫情境，皆以 FastAPI + aiofiles 為基礎。

1️⃣ 基本的非同步檔案上傳與儲存

from fastapi import FastAPI, File, UploadFile
import aiofiles
import os

app = FastAPI()
UPLOAD_DIR = "uploaded_files"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    """
    接收單一檔案，非同步寫入磁碟。
    """
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    async with aiofiles.open(file_path, "wb") as out_file:
        # 使用迭代方式分塊寫入，避免一次讀入過大記憶體
        while content := await file.read(1024 * 1024):  # 1 MB 為一塊
            await out_file.write(content)
    return {"filename": file.filename, "size": os.path.getsize(file_path)}

說明

• await file.read(size) 會以非同步方式讀取檔案內容。
• aiofiles.open(..., "wb") 以非同步方式開啟寫入檔案，await out_file.write(...) 同樣不會阻塞。

---

2️⃣ 同時上傳多個檔案

from typing import List

@app.post("/upload-multiple")
async def upload_multiple(files: List[UploadFile] = File(...)):
    """
    同時處理多個檔案，上傳時每個檔案都使用非同步寫入。
    """
    saved_files = []
    for file in files:
        file_path = os.path.join(UPLOAD_DIR, file.filename)
        async with aiofiles.open(file_path, "wb") as out_file:
            while chunk := await file.read(1024 * 1024):
                await out_file.write(chunk)
        saved_files.append({"filename": file.filename, "size": os.path.getsize(file_path)})
    return {"files": saved_files}

技巧：若同時上傳大量檔案，建議使用 asyncio.gather 把每個檔案的寫入任務併發執行，進一步提升效能（下例 4 會說明）。

---

3️⃣ 讀取已上傳的檔案並回傳部分內容（例如預覽）

@app.get("/preview/{filename}")
async def preview_file(filename: str, lines: int = 10):
    """
    讀取已儲存的文字檔，僅回傳前 `lines` 行，避免一次載入過大檔案。
    """
    file_path = os.path.join(UPLOAD_DIR, filename)
    if not os.path.isfile(file_path):
        return {"error": "File not found"}

    preview = []
    async with aiofiles.open(file_path, "r", encoding="utf-8") as f:
        async for line in f:
            preview.append(line.rstrip())
            if len(preview) >= lines:
                break
    return {"filename": filename, "preview": preview}

重點：

• async for line in f 讓我們以非同步方式逐行讀取，大檔案也不會一次佔滿記憶體。
• 透過 lines 參數讓使用者自行決定預覽的行數，提升彈性。

---

4️⃣ 使用 asyncio.gather 並發寫入多檔案（進階）

import asyncio

async def save_file(file: UploadFile):
    """單一檔案的非同步寫入函式，供 gather 使用。"""
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    async with aiofiles.open(file_path, "wb") as out_file:
        while chunk := await file.read(1024 * 1024):
            await out_file.write(chunk)
    return {"filename": file.filename, "size": os.path.getsize(file_path)}

@app.post("/upload-concurrent")
async def upload_concurrent(files: List[UploadFile] = File(...)):
    """
    使用 asyncio.gather 同時寫入多個檔案，最大化 I/O 效能。
    """
    tasks = [save_file(file) for file in files]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 處理可能的例外
    saved = []
    errors = []
    for res in results:
        if isinstance(res, Exception):
            errors.append(str(res))
        else:
            saved.append(res)

    return {"saved": saved, "errors": errors}

說明

• asyncio.gather 會同時執行所有 save_file 任務，適合 CPU 輕量、 I/O 密集的情境。
• return_exceptions=True 可捕捉單個檔案寫入失敗，而不影響其他檔案的處理。

---

常見陷阱與最佳實踐

陷阱	可能的後果	解決方式
忘記使用 await	同步執行導致阻塞，效能下降	確認所有 aiofiles 操作前都有 await
一次讀取過大區塊	記憶體使用量激增，甚至 OOM	建議每次讀取 1~4 MB，視實際需求調整
直接使用 File 接收大檔案	檔案會一次載入記憶體，造成高負載	改用 UploadFile，支援流式讀寫
未建立儲存目錄	FileNotFoundError	在程式啟動時使用 os.makedirs(..., exist_ok=True)
缺乏檔案類型驗證	安全風險（上傳惡意執行檔）	使用 UploadFile.content_type 或自訂驗證函式
同步寫入後未關閉檔案	檔案資源泄漏	使用 async with 自動關閉，或手動 await f.close()

最佳實踐

1. 使用 async with aiofiles.open：確保檔案在例外發生時仍能正確關閉。
2. 分塊讀寫：chunk_size 依硬體與網路環境調整，避免一次過大 I/O。
3. 驗證檔案類型與大小：在路由層面先檢查 file.content_type、file.size（需要自行取得），減少不必要的磁碟寫入。
4. 設定合理的上傳限制：在 FastAPI 中可透過 max_length 或使用中間件限制請求體大小。
5. 日誌與監控：使用 logging 記錄上傳成功與失敗的檔案資訊，配合 APM（如 Prometheus）觀測 I/O 延遲。

---

實際應用場景

場景	為何需要非同步檔案讀寫	可能的實作方式
圖像辨識平台（使用者上傳高解析度圖片）	圖片檔案往往超過 5 MB，若同步寫入會阻塞其他請求	使用 UploadFile + aiofiles，配合背景任務（BackgroundTasks）進行後續處理
日誌匯入系統（每日上傳數 GB CSV）	大量 CSV 必須快速寫入磁碟，同時允許多個使用者同時上傳	asyncio.gather 並發寫入，寫入完成後觸發資料庫匯入工作
影片剪輯服務（上傳影片片段）	影片檔案常在 GB 級，需即時寫入以免佔滿記憶體	直接流式寫入磁碟，寫入完成後推送至雲端儲存（如 S3）
線上表單（使用者上傳附件）	附件大小不一，且表單需要即時回應	使用 UploadFile，先寫入暫存目錄，完成驗證後搬移至正式目錄

---

總結

• 非同步檔案讀寫 是 FastAPI 在高併發環境下保持效能的關鍵。
• 透過 UploadFile 搭配 aiofiles，我們能在不阻塞事件迴圈的前提下完成大檔案的上傳、分塊寫入與流式讀取。
• 了解 分塊大小、例外處理與目錄管理 等細節，能避免常見的記憶體泄漏與 I/O 瓶頸。
• 在實務專案中，結合 asyncio.gather、背景任務與日誌監控，即可打造既安全又具伸縮性的檔案上傳服務。

掌握了上述概念與範例後，你就能在 FastAPI 中自信地處理任何規模的檔案上傳需求，為使用者提供流暢且可靠的體驗。祝開發順利！ 🚀