本文 AI 產出,尚未審核

FastAPI – 表單與檔案上傳(Form & File Upload)

主題:儲存上傳檔案到伺服器


簡介

在現代的 Web 應用程式中,檔案上傳 是最常見的需求之一,無論是使用者上傳大頭照、簡報檔,或是系統接收外部資料做後續分析,都離不開安全、快速且易於維護的上傳機制。
FastAPI 以 高效能、簡潔的程式介面 為設計核心,提供了原生支援 FormFile 的路由參數,讓開發者只需幾行程式碼即可完成檔案的接收與儲存。

本篇文章將深入探討 如何把使用者上傳的檔案安全地寫入伺服器磁碟,從基礎概念、實作範例到常見陷阱與最佳實務,協助初學者快速上手,同時為中級開發者提供可直接套用於專案的技巧。


核心概念

1. FastAPI 中的 UploadFileFile

  • UploadFile 是 FastAPI 為檔案上傳所設計的封裝類別,內含 filenamecontent_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) 風險。必須:

  1. 正規化檔名os.path.basename
  2. 限制檔案類型(檢查副檔名或 MIME)
  3. 產生唯一檔名(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 端再次驗證

最佳實踐

  1. 非同步寫入:除非確定檔案極小,否則一律使用 aiofiles
  2. 產生唯一檔名uuid4().hexhashlib.sha256 或時間戳 + 隨機字串皆可。
  3. 分層目錄:依功能或日期建立子目錄(uploads/2025/11/20/),有助於檔案管理與備份。
  4. 安全性檢查:結合 檔案類型白名單大小上限路徑正規化
  5. 日誌紀錄:記錄上傳者 IP、檔名、儲存路徑與大小,便於審計與除錯。
  6. 清除過期檔案:使用排程(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.ioResumable.js,伺服器端使用 aiofiles 寫入暫存分片,最後合併
臨時檔案儲存 上傳後只保留 24 小時 寫入 tmp/ 目錄,使用 APScheduler 每日清理過期檔案

總結

FastAPI表單與檔案上傳 提供了直觀且高效的 API,配合 UploadFileFileaiofiles,開發者可以在 毫秒級的延遲低記憶體佔用 下,安全地把檔案寫入伺服器。

本文從 概念說明多種實作範例常見陷阱最佳實踐,一路帶領讀者完成從 單檔同步寫入多檔非同步、驗證、抽象化 的完整流程。只要遵守以下要點,即可在任何 FastAPI 專案中可靠地處理檔案上傳:

  1. 使用 UploadFile 並以非同步方式寫入
  2. 正規化檔名、產生唯一檔名,防止路徑穿越與檔名衝突。
  3. 驗證檔案類型與大小,保護系統免受惡意檔案攻擊。
  4. 將儲存邏輯抽離成可重用的函式或 Dependency,提升程式碼可維護性。
  5. 配合目錄結構、日誌與清理機制,確保長期運營的穩定與安全。

掌握這些技巧後,你就能在 Web 應用、資料分析平台、媒體服務等 各種場景中,輕鬆實現可靠的檔案上傳與儲存功能。祝開發順利,期待看到你用 FastAPI 打造的精彩作品!