FastAPI 課程 – 請求生命週期（Request Lifecycle）

主題：背景任務（BackgroundTasks）

---

簡介

在 Web 應用程式中，請求（Request） 通常是同步完成的：從收到客戶端的資料、處理業務邏輯、產生回應，到最後把回應送回，都在同一條執行緒裡完成。
然而，許多情境下我們希望把 耗時 或 非即時 的工作交給系統在背景執行，讓 API 能盡快回應使用者，提升使用者體驗與系統吞吐量。

FastAPI 為此提供了 BackgroundTasks 這個輕量級工具，讓開發者可以在不離開 FastAPI 框架的前提下，簡單、可預測地排程背景工作。本文將從概念、實作範例、常見陷阱與最佳實踐，逐步說明如何在 FastAPI 中正確使用背景任務，並探討它在真實專案中的應用場景。

---

核心概念

1. 為什麼需要 BackgroundTasks？

• 即時回應：如寄送驗證信、產生報表、壓縮圖片等，若在主請求裡完成，使用者必須等待數秒甚至更久。
• 資源隔離：背景工作可以在不同執行緒或協程中執行，避免阻塞主事件迴圈（event loop）。
• 簡化架構：不必額外引入 Celery、RQ 等完整的任務佇列系統，對於中小型專案或原型開發非常友好。

注意：BackgroundTasks 並不是一個完整的分散式任務系統；它只在 同一個 FastAPI 應用程式的執行實例 內執行，若服務重啟或容器被重新部署，未完成的背景任務會被中斷。

2. BackgroundTasks 的工作原理

• FastAPI 會在 回應送出之前，把注入的 BackgroundTasks 物件加入 Response 中。
• 當 Response 被送出後，FastAPI 會呼叫 BackgroundTasks.run()，在 同一個事件迴圈（或執行緒）裡依序執行所有排入的函式。
• 每個背景函式的 參數 必須是 可序列化（JSON 可序列化的類型）或是 FastAPI 依賴注入 支援的型別。

3. 使用方式概覽

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    # 這是一個簡單的背景工作，寫入檔案或資料庫
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@app.post("/items/")
async def create_item(name: str, background_tasks: BackgroundTasks):
    # 主請求邏輯：儲存資料、回傳結果
    # ...
    # 把背景工作排入佇列
    background_tasks.add_task(write_log, f"Item created: {name}")
    return {"message": f"Item {name} created"}

• background_tasks: BackgroundTasks 會自動注入。
• add_task() 接收 函式 與 其參數，FastAPI 會在回應送出後呼叫它。

---

程式碼範例

以下提供 5 個實用範例，從最簡單的日誌寫入，到較進階的 Email 發送、檔案處理與自訂執行緒池，示範 BackgroundTasks 在不同情境下的使用方式。每段程式碼均附有說明註解，方便讀者快速掌握要點。

範例 1：寫入操作日誌（最基礎）

# file: app.py
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def log_action(action: str, item_id: int):
    """把重要操作寫入伺服器端日誌檔案。"""
    with open("actions.log", "a", encoding="utf-8") as f:
        f.write(f"{action} - item_id={item_id}\n")

@app.post("/items/{item_id}")
async def update_item(item_id: int, name: str, background_tasks: BackgroundTasks):
    # 假設這裡會寫入 DB 或其他商業邏輯
    # ...

    # 把寫檔工作交給背景任務
    background_tasks.add_task(log_action, "update", item_id)
    return {"status": "updated", "item_id": item_id}

重點：日誌寫入不需要回傳結果給使用者，放在背景執行即可避免阻塞。

---

範例 2：非同步寄送驗證信件

# file: email_utils.py
import aiosmtplib
from email.message import EmailMessage

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")

# file: main.py
from fastapi import FastAPI, BackgroundTasks
from email_utils import send_verification_email
import uuid

app = FastAPI()

@app.post("/register/")
async def register_user(email: str, background_tasks: BackgroundTasks):
    # 產生驗證 token，寫入資料庫（略）
    token = str(uuid.uuid4())

    # 把寄信工作排入背景任務（注意使用 async 函式）
    background_tasks.add_task(send_verification_email, email, token)
    return {"msg": "註冊成功，請檢查信箱驗證"}

技巧：BackgroundTasks 能接受 非同步函式（async def），FastAPI 會在事件迴圈中正確呼叫它。

---

範例 3：大量圖片壓縮（CPU 密集型）

# file: image_tasks.py
from pathlib import Path
from PIL import Image
import shutil

def compress_image(src_path: str, dst_path: str, quality: int = 70):
    """將圖片壓縮後寫入目標路徑。"""
    with Image.open(src_path) as img:
        img.save(dst_path, optimize=True, quality=quality)

def move_original(src_path: str, archive_dir: str):
    """把原始檔搬到備份資料夾。"""
    Path(archive_dir).mkdir(parents=True, exist_ok=True)
    shutil.move(src_path, f"{archive_dir}/{Path(src_path).name}")

# file: main.py
from fastapi import FastAPI, UploadFile, File, BackgroundTasks
from image_tasks import compress_image, move_original
import os

app = FastAPI()
UPLOAD_DIR = "uploads"
COMPRESSED_DIR = "compressed"
ARCHIVE_DIR = "archive"

@app.post("/upload-image/")
async def upload_image(file: UploadFile = File(...), background_tasks: BackgroundTasks = None):
    # 儲存原始檔案
    raw_path = os.path.join(UPLOAD_DIR, file.filename)
    os.makedirs(UPLOAD_DIR, exist_ok=True)
    with open(raw_path, "wb") as f:
        f.write(await file.read())

    # 壓縮檔案路徑
    compressed_path = os.path.join(COMPRESSED_DIR, file.filename)

    # 把 CPU 密集型工作交給背景任務
    background_tasks.add_task(compress_image, raw_path, compressed_path, 60)
    background_tasks.add_task(move_original, raw_path, ARCHIVE_DIR)

    return {"msg": "檔案已上傳，壓縮與搬移將在背景完成"}

說明：圖片壓縮是 CPU 密集型 工作，若使用 BackgroundTasks，仍會在同一個事件迴圈的執行緒中執行。對於大量或高頻率的壓縮需求，建議改為 執行緒池（見範例 5）或外部任務佇列。

---

範例 4：使用 Depends 注入共用背景任務物件

在大型專案中，可能會想把背景任務的設定或共用資源（例如資料庫連線、Redis 客戶端）抽象成 依賴，讓每個 endpoint 都能直接使用。

# file: dependencies.py
from fastapi import Depends, BackgroundTasks

def get_background_tasks(background_tasks: BackgroundTasks = None):
    """依賴工廠，讓其他函式可以直接取得 BackgroundTasks 實例。"""
    return background_tasks

# file: main.py
from fastapi import FastAPI, Depends
from dependencies import get_background_tasks
from typing import List

app = FastAPI()

def log_bulk_actions(actions: List[str], background_tasks: BackgroundTasks):
    """一次寫入多筆日誌，示範在依賴中使用背景任務。"""
    for act in actions:
        background_tasks.add_task(lambda a=act: print(f"[LOG] {a}"))

@app.post("/bulk-action/")
async def bulk_action(actions: List[str], bg: BackgroundTasks = Depends(get_background_tasks)):
    # 主邏輯
    # ...

    # 使用依賴注入的背景任務
    log_bulk_actions(actions, bg)
    return {"status": "queued"}

重點：透過 Depends 注入 BackgroundTasks，可以在 服務層（service layer） 中直接使用背景任務，而不必把 BackgroundTasks 參數傳遞到每一個業務函式。

---

範例 5：自訂執行緒池（ThreadPoolExecutor）結合 BackgroundTasks

如果背景工作是 CPU 密集 或 阻塞 I/O（例如呼叫外部 API、處理大型檔案），單純使用 BackgroundTasks 可能仍會阻塞事件迴圈。此時可以自行建立執行緒池，並在背景任務中提交工作。

# file: thread_pool.py
from concurrent.futures import ThreadPoolExecutor
import time

# 建立全域執行緒池（根據需求調整 max_workers）
executor = ThreadPoolExecutor(max_workers=5)

def heavy_computation(data: int) -> int:
    """模擬耗時的 CPU 工作。"""
    time.sleep(3)  # 假裝計算需要 3 秒
    return data * data

# file: main.py
from fastapi import FastAPI, BackgroundTasks
from thread_pool import executor, heavy_computation

app = FastAPI()

def schedule_heavy_task(data: int):
    """把 CPU 密集工作提交到執行緒池，並在完成後印出結果。"""
    future = executor.submit(heavy_computation, data)
    # 可選：future.add_done_callback(lambda f: print(f"Result: {f.result()}"))
    # 為了示範，我們直接在背景任務裡等待結果（非必要）
    result = future.result()
    print(f"[ThreadPool] 計算結果: {result}")

@app.get("/compute/{value}")
async def compute(value: int, background_tasks: BackgroundTasks):
    # 把耗時工作交給自訂執行緒池
    background_tasks.add_task(schedule_heavy_task, value)
    return {"msg": f"已排程計算 {value} 的平方，稍後會在伺服器端完成"}

說明：

1. ThreadPoolExecutor 允許同時執行多個阻塞任務，避免阻塞 FastAPI 的事件迴圈。
2. 若需在任務完成後回傳結果給使用者，請改用 WebSocket、Server‑Sent Events 或外部佇列（如 Redis）再通知前端。

---

常見陷阱與最佳實踐

常見陷阱	為何會發生	解決方案 / 最佳實踐
背景任務執行失敗卻沒被捕獲	BackgroundTasks 內部不會自動捕獲例外，若函式拋出錯誤，會在日誌中顯示但不會回傳給客戶端。	在背景函式內部 自行捕獲例外，或使用 try/except 並寫入錯誤日誌。
使用阻塞 I/O（例如 requests）導致事件迴圈卡住	BackgroundTasks 本身仍在同一個事件迴圈執行，阻塞呼叫會阻塞其他請求。	改用 非同步 I/O（httpx.AsyncClient）或把阻塞呼叫包在 ThreadPoolExecutor 中。
大量背景任務導致記憶體泄漏	背景任務若持有大型物件（如完整檔案內容）而未釋放，會持續佔用記憶體。	傳遞檔案路徑或唯一識別碼，在背景任務內部重新打開檔案，避免在請求中持有大物件。
服務重啟導致未完成任務遺失	BackgroundTasks 僅在單一實例內執行，容器重啟或程式崩潰時任務不會持久化。	需要 持久化任務 時，改用 Celery、RQ、或使用 Redis Streams 等外部佇列。
背景任務中使用 FastAPI 的依賴（如 DB Session）	依賴通常在請求結束時關閉，若在背景任務裡使用已關閉的資源會拋錯。	在背景任務內 重新建立 所需的資源（例如建立新的 DB Session），或使用 依賴工廠 產生臨時資源。

最佳實踐

1. 保持背景函式輕量：僅做「一次性、非即時」的工作，如寫日誌、發送通知、觸發外部服務。
2. 避免在背景任務裡直接操作請求物件：如 Request、Response、Cookies 等，因為請求已結束。
3. 使用非同步 I/O：若背景任務涉及網路或磁碟 I/O，盡可能採用 async 函式與相容的庫（httpx、aiosmtplib、aiofiles）。
4. 適當使用執行緒池：對於 CPU 密集或阻塞 I/O，將工作委派給 ThreadPoolExecutor 或 ProcessPoolExecutor，防止阻塞事件迴圈。
5. 日誌與監控：將每個背景任務的開始、結束與錯誤情況寫入結構化日誌，方便後續排錯與監控。
6. 測試：使用 TestClient 時，BackgroundTasks 會在測試結束前自動執行，確保測試涵蓋背景邏輯。

---

實際應用場景

場景	為何適合使用 BackgroundTasks	範例說明
使用者註冊後寄送驗證郵件	需要即時回應註冊成功訊息，郵件寄送可延後	範例 2 中的非同步寄信
上傳大檔案後產生縮圖	圖片壓縮耗時，使用者不必等待	範例 3 中的圖片壓縮與搬移
系統操作日誌	每筆操作都需要寫入檔案或資料庫，頻繁寫入會影響效能	範例 1 中的日誌寫入
批次資料匯入	大量資料寫入資料庫，使用者只需要知道匯入已排程	可把匯入工作放入 BackgroundTasks，或搭配執行緒池
推播通知或 Webhook	第三方服務回應時間不確定，避免阻塞主流程	在背景任務中使用 httpx.AsyncClient 呼叫外部 API

實務提示：在多容器（Kubernetes）環境中，如果背景任務的可靠性非常重要，請將重要的工作交給 外部佇列（如 Celery + RabbitMQ），而把 BackgroundTasks 留給「可接受遺失」的輕量任務。

---

總結

BackgroundTasks 是 FastAPI 提供的一個 簡潔、即插即用 的背景工作機制，讓開發者能在 保持單一程式碼基礎 的同時，將非即時、耗時的工作交給系統在回應之後自行完成。

• 它適合 日誌、Email、簡易檔案處理 等輕量任務。
• 若任務涉及 阻塞 I/O、CPU 密集 或 需要持久化，則應結合 執行緒池、非同步庫 或 專業任務佇列。

透過本文的概念說明、完整範例、常見陷阱與最佳實踐，讀者應已能在自己的 FastAPI 專案中安全、有效地使用 BackgroundTasks，提升 API 的回應速度與使用者體驗。祝開發順利，期待看到你在實務中玩出更多創意！

---