FastAPI – 表單與檔案上傳(Form & File Upload)
主題:儲存上傳檔案到伺服器
簡介
在現代的 Web 應用程式中,檔案上傳 是最常見的需求之一,無論是使用者上傳大頭照、簡報檔,或是系統接收外部資料做後續分析,都離不開安全、快速且易於維護的上傳機制。
FastAPI 以 高效能、簡潔的程式介面 為設計核心,提供了原生支援 Form 與 File 的路由參數,讓開發者只需幾行程式碼即可完成檔案的接收與儲存。
本篇文章將深入探討 如何把使用者上傳的檔案安全地寫入伺服器磁碟,從基礎概念、實作範例到常見陷阱與最佳實務,協助初學者快速上手,同時為中級開發者提供可直接套用於專案的技巧。
核心概念
1. FastAPI 中的 UploadFile 與 File
UploadFile是 FastAPI 為檔案上傳所設計的封裝類別,內含filename、content_type以及一個SpooledTemporaryFile物件,支援非同步讀寫。File(...)是一個 依賴注入(Dependency Injection) 的宣告,用來告訴 FastAPI 這個參數應該從 multipart/form-data 中的檔案欄位抓取。
重點:使用
UploadFile而非bytes,能減少記憶體佔用,特別適合處理大型檔案。
2. 非同步寫入檔案
FastAPI 完全支援 async/await,因此在儲存檔案時建議使用非同步 I/O(await file.read()、await aiofiles.open()) ,避免阻塞事件迴圈,提升整體吞吐量。
3. 安全的檔案路徑處理
直接使用使用者提供的檔名寫入磁碟會產生 路徑穿越(Path Traversal) 風險。必須:
- 正規化檔名(
os.path.basename) - 限制檔案類型(檢查副檔名或 MIME)
- 產生唯一檔名(UUID、hash)以避免檔名衝突
程式碼範例
以下範例展示從最簡單的同步寫入到完整的非同步、驗證與目錄管理的實作。每段程式碼均附上說明,方便讀者一步步跟進。
1️⃣ 最簡單的同步寫入範例
# file: main_sync.py
from fastapi import FastAPI, File, UploadFile
import os
app = FastAPI()
UPLOAD_DIR = "uploaded_files"
os.makedirs(UPLOAD_DIR, exist_ok=True)
@app.post("/upload-sync")
def upload_sync(file: UploadFile = File(...)):
# 取得安全的檔名
filename = os.path.basename(file.filename)
file_path = os.path.join(UPLOAD_DIR, filename)
# 直接以二進位寫入(同步)
with open(file_path, "wb") as buffer:
buffer.write(file.file.read())
return {"filename": filename, "size": os.path.getsize(file_path)}
說明:
file.file.read()會一次把整個檔案讀進記憶體,適合小檔案(< 5 MB)。os.makedirs(..., exist_ok=True)確保上傳目錄存在。
2️⃣ 非同步寫入(推薦)
# file: main_async.py
from fastapi import FastAPI, File, UploadFile
import os
import uuid
import aiofiles
app = FastAPI()
UPLOAD_DIR = "uploaded_files"
os.makedirs(UPLOAD_DIR, exist_ok=True)
@app.post("/upload-async")
async def upload_async(file: UploadFile = File(...)):
# 使用 UUID 產生唯一檔名,避免衝突
ext = os.path.splitext(file.filename)[1]
unique_name = f"{uuid.uuid4().hex}{ext}"
file_path = os.path.join(UPLOAD_DIR, unique_name)
# 非同步寫入檔案
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 {"original_name": file.filename, "saved_as": unique_name}
說明:
await file.read(chunk_size)以 分塊(chunk) 方式讀取,避免一次佔用過多記憶體。aiofiles為非同步檔案 I/O 套件,配合async使用可讓 FastAPI 處理更多同時請求。
3️⃣ 限制檔案類型與大小
# file: main_validation.py
from fastapi import FastAPI, File, UploadFile, HTTPException
import os, uuid, aiofiles
app = FastAPI()
UPLOAD_DIR = "uploaded_images"
os.makedirs(UPLOAD_DIR, exist_ok=True)
ALLOWED_EXT = {".png", ".jpg", ".jpeg", ".gif"}
MAX_SIZE = 5 * 1024 * 1024 # 5 MB
def validate_file(filename: str, size: int):
ext = os.path.splitext(filename)[1].lower()
if ext not in ALLOWED_EXT:
raise HTTPException(status_code=400, detail="不支援的檔案類型")
if size > MAX_SIZE:
raise HTTPException(status_code=400, detail="檔案過大,請控制在 5 MB 以內")
@app.post("/upload-image")
async def upload_image(file: UploadFile = File(...)):
# 先取得檔案大小(不會把整個檔案讀進記憶體)
size = 0
async for chunk in file.iter_chunks():
size += len(chunk)
if size > MAX_SIZE:
raise HTTPException(status_code=400, detail="檔案過大")
# 重新定位檔案指標
await file.seek(0)
validate_file(file.filename, size)
# 產生唯一檔名
unique_name = f"{uuid.uuid4().hex}{os.path.splitext(file.filename)[1]}"
dest = os.path.join(UPLOAD_DIR, unique_name)
async with aiofiles.open(dest, "wb") as out_file:
while chunk := await file.read(1024 * 1024):
await out_file.write(chunk)
return {"saved_as": unique_name, "size": size}
說明:
- 先用
iter_chunks()走訪一次檔案,計算大小並做上限檢查,之後再seek(0)回到檔頭。- 只允許圖片副檔名,防止惡意上傳執行檔。
4️⃣ 多檔案同時上傳
# file: main_multi.py
from fastapi import FastAPI, File, UploadFile
import os, uuid, aiofiles
app = FastAPI()
UPLOAD_DIR = "batch_uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)
@app.post("/upload-multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
saved = []
for file in files:
ext = os.path.splitext(file.filename)[1]
unique_name = f"{uuid.uuid4().hex}{ext}"
dest = os.path.join(UPLOAD_DIR, unique_name)
async with aiofiles.open(dest, "wb") as out_file:
while chunk := await file.read(1024 * 1024):
await out_file.write(chunk)
saved.append({"original": file.filename, "saved_as": unique_name})
return {"files": saved}
說明:
- 透過
list[UploadFile]接收多個檔案,迴圈內部仍採用非同步寫入。- 回傳每個檔案的原始名稱與儲存後的唯一名稱,方便前端呈現。
5️⃣ 使用 Depends 抽離儲存邏輯(可重用)
# file: utils.py
import os, uuid, aiofiles
from fastapi import UploadFile, HTTPException
BASE_DIR = "shared_uploads"
os.makedirs(BASE_DIR, exist_ok=True)
async def save_upload_file(upload_file: UploadFile, sub_dir: str = "") -> str:
ext = os.path.splitext(upload_file.filename)[1]
unique_name = f"{uuid.uuid4().hex}{ext}"
target_dir = os.path.join(BASE_DIR, sub_dir)
os.makedirs(target_dir, exist_ok=True)
dest_path = os.path.join(target_dir, unique_name)
async with aiofiles.open(dest_path, "wb") as out_file:
while chunk := await upload_file.read(1024 * 1024):
await out_file.write(chunk)
return unique_name
# file: main_dep.py
from fastapi import FastAPI, File, UploadFile, Depends
from utils import save_upload_file
app = FastAPI()
@app.post("/upload-doc")
async def upload_doc(file: UploadFile = File(...), saved_name: str = Depends(lambda f: save_upload_file(f, "docs"))):
# 這裡 saved_name 已經是儲存後的檔名
return {"saved_as": saved_name}
說明:
- 把儲存流程抽成可重用的 Dependency,讓不同路由只需要傳入不同的子目錄即可。
常見陷阱與最佳實踐
| 陷阱 | 為什麼會發生 | 解決方案 |
|---|---|---|
一次性讀取整個檔案 (file.file.read()) |
大檔案會佔用過多記憶體,甚至導致 OOM | 使用 分塊讀取(await file.read(chunk_size))或 iter_chunks() |
| 直接使用使用者提供的檔名 | 可能包含 ../ 造成路徑穿越,或檔名衝突 |
正規化檔名、使用 UUID/Hash 產生唯一檔名 |
| 未檢查 MIME / 副檔名 | 攻擊者可上傳惡意執行檔,執行程式碼注入 | 只允許白名單副檔名或驗證 content_type |
| 同步 I/O 阻塞事件迴圈 | 高併發時請求會被卡住,效能下降 | 使用 aiofiles 進行非同步寫入 |
| 忘記建立儲存目錄 | FileNotFoundError |
在啟動時或寫入前 確保目錄存在 (os.makedirs(..., exist_ok=True)) |
| 未限制檔案大小 | 大檔案可能耗盡磁碟空間 | 在程式內或反向代理(Nginx, Traefik)設定 上傳上限,並在 FastAPI 端再次驗證 |
最佳實踐
- 非同步寫入:除非確定檔案極小,否則一律使用
aiofiles。 - 產生唯一檔名:
uuid4().hex、hashlib.sha256或時間戳 + 隨機字串皆可。 - 分層目錄:依功能或日期建立子目錄(
uploads/2025/11/20/),有助於檔案管理與備份。 - 安全性檢查:結合 檔案類型白名單、大小上限、路徑正規化。
- 日誌紀錄:記錄上傳者 IP、檔名、儲存路徑與大小,便於審計與除錯。
- 清除過期檔案:使用排程(Celery、APScheduler)定期刪除過舊或未使用的檔案,防止磁碟被填滿。
實際應用場景
| 場景 | 需求 | 參考實作 |
|---|---|---|
| 使用者頭像上傳 | 只允許圖片、檔案尺寸 ≤ 2 MB、需要即時回傳 URL | 使用 upload_image 範例,儲存於 static/avatars/,回傳相對路徑給前端 |
| 文件管理系統 | 允許 PDF、Word、Excel,需依使用者 ID 分目錄 | 在 save_upload_file 加入子目錄 user_{user_id},並寫入資料庫紀錄 |
| 大量 CSV 匯入 | 單檔可達 100 MB,需分塊讀取並即時寫入資料庫 | 以 非同步分塊方式寫入磁碟,完成後再啟動背景任務(Celery)解析 CSV |
| 多媒體平台影片上傳 | 檔案 > 1 GB,需支援斷點續傳 | 搭配前端的 Tus.io 或 Resumable.js,伺服器端使用 aiofiles 寫入暫存分片,最後合併 |
| 臨時檔案儲存 | 上傳後只保留 24 小時 | 寫入 tmp/ 目錄,使用 APScheduler 每日清理過期檔案 |
總結
FastAPI 為 表單與檔案上傳 提供了直觀且高效的 API,配合 UploadFile、File、aiofiles,開發者可以在 毫秒級的延遲 與 低記憶體佔用 下,安全地把檔案寫入伺服器。
本文從 概念說明、多種實作範例、常見陷阱、最佳實踐,一路帶領讀者完成從 單檔同步寫入 到 多檔非同步、驗證、抽象化 的完整流程。只要遵守以下要點,即可在任何 FastAPI 專案中可靠地處理檔案上傳:
- 使用
UploadFile並以非同步方式寫入。 - 正規化檔名、產生唯一檔名,防止路徑穿越與檔名衝突。
- 驗證檔案類型與大小,保護系統免受惡意檔案攻擊。
- 將儲存邏輯抽離成可重用的函式或 Dependency,提升程式碼可維護性。
- 配合目錄結構、日誌與清理機制,確保長期運營的穩定與安全。
掌握這些技巧後,你就能在 Web 應用、資料分析平台、媒體服務等 各種場景中,輕鬆實現可靠的檔案上傳與儲存功能。祝開發順利,期待看到你用 FastAPI 打造的精彩作品!