FastAPI – 背景任務（Background Tasks）

寫入檔案 / 寄信等背景任務

---

簡介

在 Web API 中，許多操作並不需要即時回傳給使用者，例如寫入大量日誌、產生報表、寄送驗證信件或通知。若直接在路由函式內執行這類耗時工作，會造成請求被阻塞，使用者必須等待不必要的時間，甚至可能導致 502/504 錯誤。

FastAPI 內建的 BackgroundTasks 讓我們可以把這類 非同步、非即時 的工作交給背景執行，而不影響主請求的回應速度。本文將從概念說明、實作範例、常見陷阱與最佳實踐，帶你完整掌握在 FastAPI 中使用背景任務寫入檔案與寄信的技巧。

---

核心概念

1. 為什麼使用 BackgroundTasks？

• 即時回應：主請求只負責驗證、處理核心邏輯，背景任務則在回傳後自行執行。
• 資源分離：CPU/IO 密集型工作不會佔用 FastAPI 的工作執行緒（或協程），降低服務阻塞機率。
• 簡易整合：不需要額外的任務佇列（如 Celery）即可快速完成小型背景工作。

注意：BackgroundTasks 仍在同一個 Python 進程中執行，若任務非常耗時或需要重試機制，仍建議使用專門的任務佇列。

2. 基本使用方式

BackgroundTasks 是 FastAPI 提供的依賴注入類別，只要在路由函式的參數中加入 background_tasks: BackgroundTasks，即可把任務加入佇列：

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a", encoding="utf-8") as f:
        f.write(message + "\n")

@app.post("/submit")
async def submit(data: dict, background_tasks: BackgroundTasks):
    # 先回傳成功訊息
    background_tasks.add_task(write_log, f"Received: {data}")
    return {"status": "queued"}

3. 背景任務的執行時機

FastAPI 會在 回傳 Response 之後，立即在同一個事件迴圈（event loop）中呼叫 add_task 所註冊的函式。若使用 async def 的背景函式，FastAPI 會 await 它；若是同步函式，則會在執行緒池（run_in_threadpool）中執行，以避免阻塞主協程。

4. 實用範例

以下提供 5 個常見且實用的背景任務範例，包括寫檔、寄信、產生 PDF、上傳至雲端儲存、以及排程簡易清理工作。

---

4.1 寫入檔案（同步函式）

import json
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def save_to_file(data: dict, filename: str = "data.json"):
    """
    把收到的資料寫入 JSON 檔案，使用追加模式。
    """
    with open(filename, "a", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
        f.write("\n")

@app.post("/log")
async def log_data(payload: dict, background_tasks: BackgroundTasks):
    # 立即回應使用者，背景寫入檔案
    background_tasks.add_task(save_to_file, payload)
    return {"msg": "資料已排入寫入任務"}

重點：檔案寫入是 IO 密集 工作，使用同步函式即可，FastAPI 會自動把它搬到執行緒池。

---

4.2 寄送驗證信（非同步函式）

import aiosmtplib
from email.message import EmailMessage
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

async def send_verification_email(to_email: str, token: str):
    """
    使用 aiosmtplib 非同步寄送驗證信。
    """
    msg = EmailMessage()
    msg["From"] = "no-reply@example.com"
    msg["To"] = to_email
    msg["Subject"] = "驗證您的帳號"
    msg.set_content(f"請點擊以下連結驗證帳號：https://example.com/verify?token={token}")

    await aiosmtplib.send(
        msg,
        hostname="smtp.example.com",
        port=587,
        start_tls=True,
        username="smtp_user",
        password="smtp_pass",
    )

@app.post("/register")
async def register(email: str, background_tasks: BackgroundTasks):
    # 假設已產生 token
    token = "abc123def456"
    # 把寄信工作交給背景
    background_tasks.add_task(send_verification_email, email, token)
    return {"msg": "註冊成功，驗證信已發送"}

技巧：使用 非同步 SMTP（aiosmtplib）可避免額外的執行緒切換，對於大量寄信更有效率。

---

4.3 產生 PDF 報表（混合使用）

import io
from fastapi import FastAPI, BackgroundTasks, Response
from reportlab.pdfgen import canvas

app = FastAPI()

def generate_pdf(data: dict) -> bytes:
    """
    產生簡易 PDF，回傳二進位內容。
    """
    buffer = io.BytesIO()
    p = canvas.Canvas(buffer)
    p.drawString(100, 750, f"報表標題：{data.get('title', '未提供')}")
    p.drawString(100, 730, f"產生時間：{data.get('time', '未知')}")
    p.showPage()
    p.save()
    return buffer.getvalue()

def save_pdf_file(content: bytes, filename: str):
    """
    把 PDF 內容寫入磁碟，作為備份。
    """
    with open(filename, "wb") as f:
        f.write(content)

@app.post("/report")
async def create_report(payload: dict, background_tasks: BackgroundTasks):
    pdf_bytes = generate_pdf(payload)          # 同步產生 PDF
    background_tasks.add_task(save_pdf_file, pdf_bytes, "report.pdf")
    # 直接回傳 PDF 給前端
    return Response(content=pdf_bytes, media_type="application/pdf")

說明：此範例同時示範 即時回傳（PDF 給使用者）與 背景保存（寫入磁碟）的情境。

---

4.4 上傳檔案至 AWS S3（非同步 + 執行緒池）

import boto3
from fastapi import FastAPI, UploadFile, File, BackgroundTasks

app = FastAPI()
s3_client = boto3.client("s3", region_name="ap-northeast-1")

def upload_to_s3(file_bytes: bytes, bucket: str, key: str):
    """
    同步上傳檔案至 S3，放入執行緒池執行。
    """
    s3_client.put_object(Body=file_bytes, Bucket=bucket, Key=key)

@app.post("/upload")
async def upload(file: UploadFile = File(...), background_tasks: BackgroundTasks):
    content = await file.read()               # 讀取上傳檔案的二進位
    # 背景上傳至 S3，避免阻塞回應
    background_tasks.add_task(upload_to_s3, content, "my-bucket", f"uploads/{file.filename}")
    return {"msg": f"{file.filename} 已排入上傳任務"}

提示：boto3 為同步套件，FastAPI 會自動把它搬到執行緒池；若想徹底非同步，可改用 aioboto3。

---

4.5 定時清理暫存目錄（簡易排程）

import os, shutil, time
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()
TMP_DIR = "./tmp"

def clean_tmp_folder(max_age_seconds: int = 86400):
    """
    刪除超過 max_age_seconds 的暫存檔案。
    """
    now = time.time()
    for filename in os.listdir(TMP_DIR):
        path = os.path.join(TMP_DIR, filename)
        if os.path.isfile(path) and now - os.path.getmtime(path) > max_age_seconds:
            os.remove(path)

@app.on_event("startup")
async def schedule_cleanup(background_tasks: BackgroundTasks):
    """
    在應用啟動時排入一次性清理任務，之後可自行使用排程工具（如 APScheduler）觸發。
    """
    background_tasks.add_task(clean_tmp_folder)

實務：若需要 週期性 執行，建議結合 APScheduler、Celery Beat 或 Kubernetes CronJob，而非僅靠 BackgroundTasks。

---

常見陷阱與最佳實踐

陷阱	可能的影響	解決方式 / 最佳實踐
背景任務阻塞主事件迴圈	大量同步 I/O（如大量檔案寫入）會卡住所有請求	使用 run_in_threadpool（FastAPI 內部自動處理）或自行將任務包裝成 async def 並使用 await
例外未被捕捉	背景任務拋出例外會被 silently swallow，難以追蹤	在背景函式最外層使用 try/except，並記錄至日誌或上傳至監控平台
共享全域變數	多執行緒/協程同時寫入同一變數會產生 race condition	使用 thread‑safe 結構（如 queue.Queue）或將資料寫入外部儲存（DB、Redis）
資源未釋放	開啟檔案、資料庫連線未關閉會導致資源泄漏	使用 with 語句或在背景函式結束前明確呼叫 close()
背景任務過多	同時排入大量任務會耗盡執行緒池，導致新請求被阻塞	設定合理的 max_workers（uvicorn/gunicorn），或改用專業的任務佇列（Celery、RQ）
長時間任務	超過數十秒的任務仍會佔用工作執行緒，影響吞吐量	把長時間工作拆分為多個短任務，或交給外部排程系統（Celery、Airflow）

最佳實踐總結：

1. 盡量保持背景任務短小（< 5 秒），超過的交給佇列系統。
2. 捕捉例外並寫入日誌，避免靜默失敗。
3. 使用 async 函式配合非同步套件（如 aiosmtplib、aioboto3），提升效能。
4. 測試佈署環境的執行緒數，確保不會因背景任務耗盡資源。
5. 將狀態或結果寫入可查詢的儲存（資料庫、Redis），讓前端或管理員可以追蹤任務進度。

---

實際應用場景

場景	為何使用 BackgroundTasks
使用者上傳圖片後即時產生縮圖	產生縮圖耗時，可在回傳成功訊息後於背景完成，前端再以輪詢或 WebSocket 取得結果。
註冊流程寄送驗證信	使用者不必等待郵件發送完成，即可看到「註冊成功」的回應，提高體驗。
每日報表產生與郵件推送	觸發 API 時立即回覆「報表排程中」，報表生成與寄送交給背景任務完成。
大量日誌寫入或審計紀錄	把每筆請求的審計資訊寫入檔案或資料庫，避免同步寫入造成瓶頸。
檔案上傳後自動備份至雲端	上傳至本機後立即回應，背景把檔案同步至 S3、Azure Blob，確保資料安全。

---

總結

• BackgroundTasks 為 FastAPI 提供一個輕量級、即插即用的背景執行機制，適合 寫入檔案、寄送信件、上傳雲端、產生報表 等 IO 密集且不需即時回傳的工作。
• 使用時只要在路由函式加入 background_tasks: BackgroundTasks，呼叫 add_task(func, *args, **kwargs) 即可；FastAPI 會在回傳 Response 後自動在同一事件迴圈或執行緒池中執行任務。
• 保持任務短小、捕捉例外、適當使用非同步套件，是避免阻塞與資源泄漏的關鍵。
• 若任務變得 長時間或需重試/排程，建議升級至 Celery、RQ、Airflow 等專業佇列系統。

掌握了背景任務的正確使用方式，你的 FastAPI 服務將能在 高併發、低延遲 的同時，完成更多「背後」的工作，提供更佳的使用者體驗與系統彈性。祝開發順利，快去試試看吧！