FastAPI 與外部服務整合 – Cloud Storage（S3、GCS）

簡介

在現代的 Web 應用程式中，檔案的上傳、下載與長期保存往往交給 雲端物件儲存（Object Storage）來處理。Amazon S3 與 Google Cloud Storage（GCS）是兩大最常被採用的服務，它們提供高可用、彈性、且具成本效益的儲存解決方案。

將 FastAPI 與這些服務結合，不僅能讓 API 直接支援大型檔案的傳輸，還能把檔案的安全性、授權、生命週期管理交給雲端平台處理，讓後端程式碼保持簡潔、易維護。本文將一步步說明如何在 FastAPI 中整合 S3 與 GCS，並提供實作範例、常見陷阱與最佳實踐，讓你快速上手。

---

核心概念

1. 為什麼要使用雲端儲存？

• 彈性伸縮：不需要自行管理磁碟空間，隨時可上傳 TB 級別的檔案。
• 高可用性：S3 / GCS 內建多區域冗餘，資料可靠度高達 99.999999999%。
• 安全與授權：支援 IAM、Bucket Policy、Pre‑signed URL 等細緻的存取控制。

重點：在設計 API 時，應把檔案本身的傳輸交給雲端儲存，僅在 FastAPI 中處理檔案的「元資料」與「授權」邏輯。

2. 主要操作

操作	S3 常用 SDK	GCS 常用 SDK
上傳檔案	boto3.client('s3').upload_fileobj()	storage.Client().bucket().blob().upload_from_file()
下載檔案	download_fileobj()	download_to_file()
產生預簽名 URL	generate_presigned_url()	generate_signed_url()
刪除物件	delete_object()	blob.delete()

3. FastAPI 中的檔案接收與回傳

FastAPI 透過 UploadFile 與 File 參數自動處理 multipart/form-data，提供 streaming 的檔案物件，適合直接傳給雲端 SDK，避免將整個檔案讀入記憶體。

from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    # `file.file` 為一個類似 io.BytesIO 的檔案流
    return {"filename": file.filename}

---

程式碼範例

以下示範 5 個常見的實務需求，分別針對 Amazon S3 與 Google Cloud Storage 實作。所有範例均以 Python 3.9+、FastAPI 為基礎。

1️⃣ 初始化 S3 客戶端（使用環境變數或 IAM Role）

# file: s3_client.py
import os
import boto3
from botocore.exceptions import ClientError

def get_s3_client():
    """
    依賴以下環境變數：
    - AWS_ACCESS_KEY_ID
    - AWS_SECRET_ACCESS_KEY
    - AWS_REGION
    若在 EC2/ECS/EKS 上執行，會自動使用 IAM Role。
    """
    return boto3.client(
        "s3",
        region_name=os.getenv("AWS_REGION", "us-east-1")
    )

2️⃣ 上傳檔案至 S3 並回傳檔案 URL

# file: main.py
from fastapi import FastAPI, UploadFile, File, HTTPException
from s3_client import get_s3_client
import uuid

app = FastAPI()
s3 = get_s3_client()
BUCKET_NAME = "my-fastapi-bucket"

@app.post("/s3/upload")
async def upload_to_s3(file: UploadFile = File(...)):
    # 產生唯一的檔名，避免衝突
    key = f"{uuid.uuid4()}_{file.filename}"
    try:
        # 直接把 FastAPI 的檔案流傳給 boto3
        s3.upload_fileobj(file.file, BUCKET_NAME, key)
    except ClientError as e:
        raise HTTPException(status_code=500, detail=str(e))

    # 產生公開 URL（若 bucket 為 private，請改用 presigned URL）
    url = f"https://{BUCKET_NAME}.s3.amazonaws.com/{key}"
    return {"filename": file.filename, "url": url}

3️⃣ 產生 S3 Pre‑signed URL（安全的下載方式）

@app.get("/s3/download-url/{key}")
def get_presigned_url(key: str):
    try:
        url = s3.generate_presigned_url(
            ClientMethod="get_object",
            Params={"Bucket": BUCKET_NAME, "Key": key},
            ExpiresIn=3600,          # 有效期 1 小時
        )
    except ClientError as e:
        raise HTTPException(status_code=404, detail="Object not found")
    return {"download_url": url}

4️⃣ 初始化 GCS 客戶端 & 上傳檔案

# file: gcs_client.py
from google.cloud import storage
import os

def get_gcs_bucket():
    """
    需要設定環境變數 GOOGLE_APPLICATION_CREDENTIALS
    指向服務帳號金鑰 JSON 檔案，或在 GKE/GCE 上使用預設帳號。
    """
    client = storage.Client()
    bucket_name = os.getenv("GCS_BUCKET", "my-fastapi-gcs")
    return client.bucket(bucket_name)

# file: main.py (續)
from gcs_client import get_gcs_bucket

gcs_bucket = get_gcs_bucket()

@app.post("/gcs/upload")
async def upload_to_gcs(file: UploadFile = File(...)):
    blob = gcs_bucket.blob(f"{uuid.uuid4()}_{file.filename}")
    # 直接把檔案流上傳，避免佔用記憶體
    blob.upload_from_file(file.file, content_type=file.content_type)
    # 取得公開 URL（視 bucket 設定而定）
    url = f"https://storage.googleapis.com/{gcs_bucket.name}/{blob.name}"
    return {"filename": file.filename, "url": url}

5️⃣ 產生 GCS Signed URL（供前端安全下載）

@app.get("/gcs/download-url/{blob_name}")
def get_gcs_signed_url(blob_name: str):
    blob = gcs_bucket.blob(blob_name)
    try:
        url = blob.generate_signed_url(
            version="v4",
            expiration=3600,               # 1 小時
            method="GET",
        )
    except Exception as e:
        raise HTTPException(status_code=404, detail=str(e))
    return {"download_url": url}

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
檔案一次讀入記憶體	直接 await file.read() 會把整個檔案載入 RAM，對大檔案會 OOM。	使用 UploadFile.file 的 stream，配合 upload_fileobj 或 upload_from_file。
Bucket 權限設定不當	設為 public-read 會造成檔案外泄；設定過於嚴格則 API 無法存取。	以 IAM Role 或 服務帳號 為主，僅在需要時產生 pre‑signed/signed URL。
未設定有效期限的 Signed URL	產生無期限的 URL 會增加資安風險。	必須 為 Signed URL 設定 ExpiresIn / expiration，建議不超過 1–2 小時。
缺少檔案類型驗證	允許上傳任意檔案類型可能導致惡意檔案執行。	在 FastAPI 中使用 file.content_type 或自行檢查副檔名、MIME。
多執行緒/非同步衝突	boto3、google‑cloud‑storage 本身是同步庫，直接在 async endpoint 中呼叫會阻塞。	使用 run_in_threadpool（FastAPI 提供）或改用 aioboto3、gcsfs 等 async 客戶端。

最佳實踐

1. 環境變數管理：使用 python-dotenv 或 Cloud Secret Manager，避免硬編碼金鑰。
2. 統一錯誤回傳：將 Cloud SDK 的例外轉為 FastAPI 的 HTTPException，保持 API 介面一致。
3. 日誌與監控：在上傳/下載成功或失敗時寫入結構化日誌，搭配 CloudWatch 或 Stackdriver 觀察流量。
4. 測試：利用 moto（模擬 S3）或 google-cloud-storage 的測試套件，撰寫單元測試，確保上傳/下載流程不會因環境變化斷裂。

---

實際應用場景

場景	需求	方案
使用者上傳大圖 / 影片	前端直接將檔案送至 API，API 必須快速回傳可直接存取的 URL。	使用 pre‑signed URL 讓前端直接上傳至 S3/GCS，API 只負責產生 URL 與紀錄 metadata。
報表或備份檔案的定期匯出	每日產生 CSV，需存放於雲端且保留 30 天。	FastAPI 背景工作（如 Celery）產生檔案後，呼叫 upload_fileobj 上傳至指定 bucket，設定 Lifecycle Policy 自動刪除過期檔案。
多租戶 SaaS 平台	每個租戶只能存取自己資料夾的檔案。	在 bucket 中以 tenant_id/ 作為前綴，並在 IAM Policy 或 Signed URL 中限制 KeyPrefix。
文件審核流程	上傳的檔案需先經過 AI 文字辨識或惡意檔案掃描，再提供下載。	API 接收檔案 → 暫存於 private bucket → 觸發 Cloud Function / Cloud Run 處理 → 處理完畢後搬移至 public bucket 或產生 Signed URL。

---

總結

• FastAPI 的 UploadFile 與 S3/GCS 的 SDK 天生相容，能以 streaming 方式安全上傳大型檔案。
• Pre‑signed / Signed URL 是最安全、最彈性的下載方式，避免將 bucket 設為公開。
• 注意 權限、有效期限、記憶體使用 等常見陷阱，並遵循 環境變數、日誌、測試 的最佳實踐。
• 透過上述範例，你可以快速在 FastAPI 中實作 檔案上傳、下載、授權，並將雲端儲存的彈性與安全性完整帶入你的應用程式。

現在就把這些程式碼搬到你的專案，讓 FastAPI 與 Cloud Storage 成為你的強大後端基礎吧！ 🚀