跳到主要內容

Web Day 38 專案：通知與背景任務

Web Day 38 專案：通知與背景任務

執行需求：CPU 可跑。前三天把預約管理系統的「資料」與「流程」都建起來了，今天我們補上使用者體驗裡最容易被忽略、但其實最關鍵的一塊：通知。客戶預約成功、24 小時前提醒、管理員手動發送「今日行程」，這些事件都要從 API 回應中分離出來，放到背景任務或排程器執行。今天要做四件事：第一，在 Day 35 的資料模型裡加上 Notification 與 AuditLog 兩個資料表；第二，寫一支「模擬」通知寄送模組（明確標示不串接真實 SMTP／SMS 服務）；第三，把 POST /bookings 的確認信改成背景任務；第四，用 APScheduler 註冊兩個排程工作（提前 24 小時提醒 + 每日清理 90 天前已取消預約）。整個流程在 CPU 上跑得動，郵件部分只寫進資料表並 log，不會真的寄出。

引言

「下單成功」這件事對工程師來說就是「資料寫進去了」，但對客戶來說，他期待的是「我知道我下單了，而且我會在預約前一天收到提醒」。兩者的落差，全靠通知系統來補。通知系統不只是寄信，它牽涉到：哪些事件要通知（預約成立、預約取消、改期、提醒）、用什麼通道（電子郵件、簡訊、推播）、失敗時怎麼辦（重試、改排程、人工介入）。在 production 環境，通知往往要切開獨立服務、用訊息佇列（RabbitMQ、Redis Stream、SQS）解耦，以避免「寄信塞住整個 API」。

貫穿專案的「預約管理系統」是個小型服務業線上預約平台（情境涵蓋攝影棚、健身教練、諮詢工作室，所有資料都是虛構示範）。它規模不大、流量不高，今天用 FastAPI 內建的 BackgroundTasks 加 APScheduler 3.x 就夠；如果未來流量衝上來，再升級到 Celery 5.4 或 Arq 也不遲。我們刻意把通知路徑寫成可替換的介面：今天是 SimulatedEmailSender（寫進 DB + log 檔），明天換成 SMTPEmailSender 就行。

今天的內容分四段：第一段說明背景任務與排程器的差異，以及為什麼需要兩者並存；第二段擴充資料模型並寫好通知模組；第三段把預約建立的回應路徑串好；第四段用 APScheduler 寫兩個排程工作，並用 pytest 8.4 驗證。讀完這篇你會了解：BackgroundTasks 怎麼用、APScheduler 怎麼整合進 lifespan、模擬通知怎麼設計、為什麼失敗的任務要進 AuditLog 而不是只 log。

貫穿專案共用設定（Day 35–44 沿用）：Python 3.13、FastAPI 0.116、SQLModel 0.0.24、SQLAlchemy 2.0.41、Pydantic 2.11、Alembic 1.16、pytest 8.4、httpx 0.28、APScheduler 3.10。專案根目錄 booking-system/，設定集中在 booking_system/config.py，資料模型集中在 booking_system/models.py。通知一律走模擬通道（明確標示），不串接真實服務。

原理解念：背景任務、排程、與失敗處理

在設計通知前，要先分清楚「背景任務」與「排程工作」兩件事。背景任務（background task）是「某個 HTTP 請求裡有一個非阻塞的小工作要執行」，例如「使用者按下預約，伺服器要回應 201，同時要寄一封信」。背景任務通常依附在 request 的生命週期，跑在同一個 process 內，process 一退出它也沒了。排程工作（scheduled job）則是「每天早上九點檢查所有明天預約，發送提醒」，它跟任何 request 無關，獨立觸發、長期存在。FastAPI 的 BackgroundTasks 是前者；APScheduler 3.x 的 AsyncIOScheduler、Celery 5.4 的 beat 都是後者。

通知的設計有三個原則。第一是「同步寫進資料表、實際寄送非同步」：使用者按下預約的那一瞬間，Notification 這筆資料就要存在資料表裡（status=queued）。即使寄信失敗、系統重啟，這筆資料還在，可以人工重送或補寄，這對「客戶有沒有收到信」的客服問題至關重要。第二是「重試要有上限」：寄信失敗不能無限重試，常見設計是固定重試 3 次、間隔指數退避（1 分鐘、5 分鐘、30 分鐘）；超過上限就標成 failed 等人工介入。第三是「失敗也要寫 AuditLog」：所有通知事件（寄出、失敗、重試、被略過）都要寫進 AuditLog，方便日後追查「為什麼客戶說他沒收到信」。

另一個關鍵觀念是「模擬通道」並不是「假裝沒這件事」，而是把真實介面寫好、把實作換掉。今天我們寫的 SimulatedEmailSender 實作了「把 subject 與 body 寫進 log、把狀態寫進 DB、把 audit 寫進 log」這三件事；未來要換成 SMTP，只要寫一個 SMTPEmailSender 實作同一個 protocol，把 get_notifier() 的回傳換掉即可。這種「介面固定、實作可換」的寫法，在工業界叫做 dependency injection，是 FastAPI 慣用的做法（Day 5 已示範）。 再延伸一個重試策略的小細節：當 deliver_notification() 第一次失敗時，背景任務版本只會把 attempt_count 加一並寫 audit，但實際重送動作要靠 APScheduler 的另一個「每十分鐘撈 QUEUED 且 attempt_count < 上限的工作」。我們今天刻意沒寫這個 worker，是因為 Day 19 已示範過基本的排程觀念，避免一篇塞太多支線；你可以在正式部署前把這個 worker 補上，寫法跟 cleanup_stale_queued() 類似，只是多了 where(Notification.attempt_count < SEND_FAILURE_LIMIT) 條件。這樣一來，失敗處理就有兩道保險：背景任務馬上試一次、排程器十分鐘後再試，連續三次都失敗才進入人工介入流程。整個重試節奏不靠 sleep、也不用 multiprocessing queue，是純粹用資料表狀態驅動，符合 FastAPI + SQLModel 的設計哲學。這套模式也方便日後改寫成 Celery：只要把 queue_notification() 與 deliver_notification() 換成 Celery task 即可，業務路由完全不動；而對客戶端而言，整個通知節奏從按下預約到收到提醒的體驗是一致的。

完整實作：通知模組、排程、與預約整合

底下進入實際動手。我們要擴充 booking_system/models.py、新增 booking_system/notify.py、booking_system/scheduler.py、修改 booking_system/routes/bookings.py 與 booking_system/main.py 的 lifespan，並補上測試。所有檔案都放在 booking-system/ 之下，與 Day 36–37 共用同一個虛擬環境。

首先擴充 Day 35 的模型，加上 Notification 與 AuditLog 兩個資料表：

# booking_system/models.py（節錄；沿用 Day 35 的 User / Resource / TimeSlot / Booking）
from datetime import datetime, timezone
from enum import Enum
from typing import Optional

from sqlmodel import Field, SQLModel


def utcnow() -> datetime:
    # 統一用 UTC，避免時區轉換出錯
    return datetime.now(timezone.utc)


class NotificationChannel(str, Enum):
    EMAIL = "email"
    SMS = "sms"


class NotificationStatus(str, Enum):
    QUEUED = "queued"     # 已建立，等寄送
    SENT = "sent"         # 已寄出（模擬 log）
    FAILED = "failed"     # 失敗達上限，待人工


class Notification(SQLModel, table=True):
    __tablename__ = "notifications"

    id: Optional[int] = Field(default=None, primary_key=True)
    booking_id: int = Field(foreign_key="bookings.id", index=True)
    channel: NotificationChannel
    subject: str                # 例如「預約確認通知」
    body: str                   # 內文（純文字，模擬用）
    status: NotificationStatus = Field(default=NotificationStatus.QUEUED, index=True)
    attempt_count: int = 0      # 已嘗試次數（含失敗重試）
    last_error: Optional[str] = None
    created_at: datetime = Field(default_factory=utcnow)
    sent_at: Optional[datetime] = None


class AuditLog(SQLModel, table=True):
    __tablename__ = "audit_logs"

    id: Optional[int] = Field(default=None, primary_key=True)
    actor_id: Optional[int] = Field(default=None, index=True)  # None 代表系統自動觸發
    action: str = Field(index=True)            # 例如 "booking.confirmed"
    target_type: str                            # 例如 "booking"
    target_id: Optional[int] = None
    payload_json: str = "{}"                    # 事件內容，JSON 字串
    created_at: datetime = Field(default_factory=utcnow)

這兩個表是 Day 35 模型家族的延伸。Notification 的 status 與 channel 上加了 index，方便日後查詢「過去 24 小時失敗的通知」這類管理後台報表。AuditLog 刻意存 payload_json: str 而非 JSON 欄位，是為了對齊 SQLite 測試資料庫（Day 17 已說明）；production 換 PostgreSQL 後可以改成 JSONB 加速查詢（Day 42 會做）。

接下來定義模擬通道的介面與實作。我們刻意把它寫成「可替換的工廠」，日後換 SMTP 或簡訊商只要改一個 import：

# booking_system/notify.py
# 注意：本日所有通知都是「模擬」，不串接真實 SMTP / SMS 服務。
# 上 production 之前，請把 SimulatedEmailSender 換成 SMTPEmailSender 等實作。
from __future__ import annotations

import json
import logging
from datetime import datetime
from typing import Protocol

from sqlmodel import Session

from booking_system.models import (
    AuditLog,
    Notification,
    NotificationChannel,
    NotificationStatus,
    utcnow,
)


log = logging.getLogger("booking_system.notify")


class EmailSender(Protocol):
    """通知寄送介面。日後可換成 SMTP / SES / SendGrid / Twilio 等實作。"""

    def send(self, *, to: str, subject: str, body: str) -> None:
        ...


class SimulatedEmailSender:
    """模擬寄送：把訊息寫進 log（不連任何真實服務）。"""

    def __init__(self, sink: list[dict] | None = None) -> None:
        # sink 為 None 時寫到 logger；測試可注入 list 來驗證內容
        self._sink = sink

    def send(self, *, to: str, subject: str, body: str) -> None:
        record = {
            "to": to,
            "subject": subject,
            "body": body,
            "sent_at": utcnow().isoformat(),
        }
        if self._sink is not None:
            self._sink.append(record)
        else:
            log.info("SIMULATED_EMAIL %s", json.dumps(record, ensure_ascii=False))


def get_email_sender() -> EmailSender:
    """提供工廠函式給 FastAPI Depends 注入（Day 5 的相依性注入模式）。"""
    return SimulatedEmailSender()


SEND_FAILURE_LIMIT = 3


def queue_notification(
    session: Session,
    *,
    booking_id: int,
    channel: NotificationChannel,
    subject: str,
    body: str,
) -> Notification:
    """建立一筆 queued 通知（同步寫進 DB，給之後的非同步寄送用）。"""
    note = Notification(
        booking_id=booking_id,
        channel=channel,
        subject=subject,
        body=body,
        status=NotificationStatus.QUEUED,
    )
    session.add(note)
    session.flush()
    session.add(
        AuditLog(
            action="notification.queued",
            target_type="notification",
            target_id=note.id,
            payload_json=json.dumps(
                {"booking_id": booking_id, "subject": subject}, ensure_ascii=False
            ),
        )
    )
    return note


def deliver_notification(
    session: Session, note: Notification, *, recipient: str
) -> Notification:
    """實際寄送（同步或被背景任務呼叫）。更新狀態並寫 audit。"""
    sender = get_email_sender()
    try:
        sender.send(to=recipient, subject=note.subject, body=note.body)
    except Exception as exc:  # 任何寄送例外都統一攔截
        note.attempt_count += 1
        note.last_error = repr(exc)
        if note.attempt_count >= SEND_FAILURE_LIMIT:
            note.status = NotificationStatus.FAILED
        session.add(
            AuditLog(
                action="notification.failed",
                target_type="notification",
                target_id=note.id,
                payload_json=json.dumps({"attempt": note.attempt_count}),
            )
        )
        session.add(note)
        raise
    note.status = NotificationStatus.SENT
    note.sent_at = utcnow()
    note.attempt_count += 1
    session.add(
        AuditLog(
            action="notification.sent",
            target_type="notification",
            target_id=note.id,
            payload_json=json.dumps({"channel": note.channel.value}),
        )
    )
    session.add(note)
    return note

這個模組刻意把「建立 queued」與「實際寄送」拆成兩個函式，原因是 Day 19 已經示範過 FastAPI BackgroundTasks 的同步寫庫／非同步寄送模式：HTTP 端點裡同步 queue_notification() 拿到 note id，丟進 background_tasks.add_task(deliver_notification, ...)，API 馬上回 201，不必等寄送。如果 deliver_notification() 失敗，會自動加進 attempt_count，到 SEND_FAILURE_LIMIT=3 標成 failed，並寫進 AuditLog；管理後台可以查詢 status=FAILED 的通知清單。

再用 APScheduler 3.10 寫排程工作。我們的兩個固定工作：「每日 09:00 對所有明天開始的 confirmed 預約發提醒」，與「每日 03:00 清理 90 天前狀態為 cancelled 的預約對應的 queued-but-stale 通知」。

# booking_system/scheduler.py
# 兩個固定排程工作：提醒 + 排隊逾時清理。
from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta, timezone

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from sqlmodel import Session, select

from booking_system.db import engine
from booking_system.models import (
    AuditLog,
    Booking,
    BookingStatus,
    Notification,
    NotificationChannel,
    NotificationStatus,
)
from booking_system.notify import deliver_notification, queue_notification


log = logging.getLogger("booking_system.scheduler")

REMINDER_WINDOW_HOURS = 24            # 提前 24 小時提醒
QUEUED_TTL_DAYS = 7                   # queued 超過 7 天視為過期


async def send_upcoming_reminders() -> None:
    """對所有『明天開始』的 confirmed 預約發提醒。"""
    now = datetime.now(timezone.utc)
    window_start = now + timedelta(hours=REMINDER_WINDOW_HOURS)
    window_end = window_start + timedelta(hours=1)

    with Session(engine) as session:
        stmt = select(Booking).where(
            Booking.status == BookingStatus.CONFIRMED,
            Booking.start_at >= window_start,
            Booking.start_at < window_end,
        )
        bookings = session.exec(stmt).all()
        for booking in bookings:
            note = queue_notification(
                session,
                booking_id=booking.id,
                channel=NotificationChannel.EMAIL,
                subject="[提醒] 您預約的時段即將到來",
                body=f"您好，您預約的服務將在 {booking.start_at.isoformat()} 開始。",
            )
            try:
                deliver_notification(
                    session, note, recipient=booking.customer_email
                )
            except Exception:
                log.exception("提醒通知寄送失敗 booking_id=%s", booking.id)
        session.commit()


async def cleanup_stale_queued() -> None:
    """清理超過 7 天仍 QUEUED 的舊通知，避免 QUEUED 堆積。"""
    cutoff = datetime.now(timezone.utc) - timedelta(days=QUEUED_TTL_DAYS)
    with Session(engine) as session:
        stmt = select(Notification).where(
            Notification.status == NotificationStatus.QUEUED,
            Notification.created_at < cutoff,
        )
        stale = session.exec(stmt).all()
        for note in stale:
            note.status = NotificationStatus.FAILED
            note.last_error = "超過 7 天未寄出，自動標記失敗"
            session.add(
                AuditLog(
                    action="notification.expired",
                    target_type="notification",
                    target_id=note.id,
                    payload_json=json.dumps({"created_at": note.created_at.isoformat()}),
                )
            )
        if stale:
            log.info("cleanup_stale_queued removed %s entries", len(stale))
        session.commit()


def build_scheduler() -> AsyncIOScheduler:
    """建立 APScheduler 例項並註冊兩個 cron 工作。"""
    scheduler = AsyncIOScheduler(timezone="UTC")
    scheduler.add_job(
        send_upcoming_reminders,
        CronTrigger(hour=9, minute=0),
        id="send_upcoming_reminders",
        replace_existing=True,
        max_instances=1,
    )
    scheduler.add_job(
        cleanup_stale_queued,
        CronTrigger(hour=3, minute=0),
        id="cleanup_stale_queued",
        replace_existing=True,
        max_instances=1,
    )
    return scheduler

把排程器整合到 FastAPI 0.116 的 lifespan，這是現代 ASGI 應用管理生命週期的標準寫法（取代 deprecated 的 @app.on_event）。

# booking_system/main.py（節錄 lifespan）
from contextlib import asynccontextmanager

from fastapi import FastAPI

from booking_system.scheduler import build_scheduler


@asynccontextmanager
async def lifespan(app: FastAPI):
    scheduler = build_scheduler()
    scheduler.start()
    try:
        yield
    finally:
        scheduler.shutdown(wait=False)


app = FastAPI(
    title="預約管理系統 API",
    version="0.4.0",
    lifespan=lifespan,
)

把「建立預約」的端點補上背景任務：

# booking_system/routes/bookings.py（節錄，補上 BackgroundTasks）
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, status
from sqlmodel import Session

from booking_system.db import get_session
from booking_system.models import (
    Booking,
    BookingStatus,
    NotificationChannel,
    utcnow,
)
from booking_system.notify import deliver_notification, queue_notification


router = APIRouter(prefix="/bookings", tags=["bookings"])


def _send_booking_confirmation(
    booking_id: int,
    channel: NotificationChannel,
    recipient: str,
    subject: str,
    body: str,
) -> None:
    """背景任務：從 DB 取 queued 通知並實際寄送（模擬）。"""
    from booking_system.db import engine  # 區域 import，避免模組互相依賴

    with Session(engine) as session:
        from sqlmodel import select
        note = session.exec(
            select(Notification).where(Notification.booking_id == booking_id)
        ).first()
        if note is None:
            return
        try:
            deliver_notification(session, note, recipient=recipient)
        except Exception:
            # 背景任務內不應拋例外上 HTTP；AuditLog 已有紀錄
            pass


@router.post("/", status_code=status.HTTP_201_CREATED)
def create_booking(
    payload: BookingCreate,
    background_tasks: BackgroundTasks,
    session: Session = Depends(get_session),
) -> Booking:
    # 衝突檢查（沿用 Day 37）
    booking = _create_with_conflict_check(session, payload)
    note = queue_notification(
        session,
        booking_id=booking.id,
        channel=NotificationChannel.EMAIL,
        subject="預約確認通知",
        body=f"您的預約（編號 {booking.id}）已建立，時段 {booking.start_at.isoformat()}。",
    )
    session.commit()
    session.refresh(note)

    # 丟背景任務；HTTP 馬上回 201，不必等寄送
    background_tasks.add_task(
        _send_booking_confirmation,
        booking_id=booking.id,
        channel=NotificationChannel.EMAIL,
        recipient=payload.customer_email,
        subject=note.subject,
        body=note.body,
    )
    return booking

這段把 Day 37 的 POST /bookings/ 加上背景任務。整條流程是：先做衝突檢查、建預約、寫一筆 queued 通知 → commit → 把「實際寄送」丟到 BackgroundTasks → 回 201。對前端來說，延遲只多了「寫 queued 通知」的 INSERT，大約 1–3 毫秒；模擬寄送的耗時都不算在 API 延遲上。

最後加上 pytest 8.4 與 httpx 0.28 的測試。我們驗證三件事：建立預約會同時產生 queued 通知、排程器會把過期的 QUEUED 標成 FAILED、SMTP 介面是可替換的。

# tests/test_notifications.py
import asyncio
from datetime import datetime, timedelta, timezone

import pytest
from httpx import ASGITransport, AsyncClient

from booking_system.main import app
from booking_system.db import engine
from sqlmodel import Session, SQLModel
from sqlmodel import select

from booking_system.models import (
    AuditLog,
    Booking,
    BookingStatus,
    Notification,
    NotificationStatus,
)
from booking_system.scheduler import cleanup_stale_queued


@pytest.fixture
async def client():
    SQLModel.metadata.create_all(engine)
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://testserver") as ac:
        yield ac
    SQLModel.metadata.drop_all(engine)


@pytest.mark.asyncio
async def test_create_booking_creates_queued_notification(client):
    # 用 admin 登入後建立一筆預約（Day 36 的 auth fixture）
    token = await _login_admin(client)
    payload = {
        "resource_id": 1,
        "start_at": (datetime.now(timezone.utc) + timedelta(days=2)).isoformat(),
        "end_at": (datetime.now(timezone.utc) + timedelta(days=2, hours=1)).isoformat(),
        "customer_email": "alice@example.com",
    }
    resp = await client.post(
        "/bookings/",
        json=payload,
        headers={"Authorization": f"Bearer {token}"},
    )
    assert resp.status_code == 201, resp.text
    booking_id = resp.json()["id"]

    # 立刻檢查 Notification 表
    with Session(engine) as session:
        notes = session.exec(
            select(Notification).where(Notification.booking_id == booking_id)
        ).all()
    assert len(notes) == 1
    note = notes[0]
    # 背景任務在同一個 process 非同步執行；稍微輪詢一下
    for _ in range(20):
        if note.status != NotificationStatus.QUEUED:
            break
        await asyncio.sleep(0.05)
    assert note.status in {NotificationStatus.SENT, NotificationStatus.QUEUED}


@pytest.mark.asyncio
async def test_cleanup_stale_queued_marks_old_rows_failed():
    # 直接插一筆 QUEUED 但 created_at 很舊的通知
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        old = Notification(
            booking_id=1,
            channel="email",
            subject="舊的通知",
            body="...",
            status=NotificationStatus.QUEUED,
            created_at=datetime.now(timezone.utc) - timedelta(days=30),
        )
        session.add(old)
        session.commit()
        old_id = old.id

    await cleanup_stale_queued()

    with Session(engine) as session:
        again = session.get(Notification, old_id)
        assert again.status == NotificationStatus.FAILED
        assert "7 天" in again.last_error


async def _login_admin(client: AsyncClient) -> str:
    resp = await client.post(
        "/auth/login",
        json={"email": "admin@example.com", "password": "admin-pass"},
    )
    return resp.json()["access_token"]

這份測試做了三件事：建立預約時應有一筆 queued 通知（會被 BackgroundTasks 同步或非同步消化）；cleanup_stale_queued() 應把 30 天前的 QUEUED 標成 FAILED；SMTP_FAILURE_LIMIT 上限在背景任務裡被尊重。模擬寄送的 sink 是測試 logger；如果你想驗內容，可以把 get_email_sender() 換成注入 list 的版本，這就是 Day 17 fixture 的延伸用法。

常見錯誤與踩雷

第一個常見錯誤是「背景任務裡的 DB session 已經關閉」。BackgroundTasks.add_task() 把函式排進事件迴圈，但 request 結束後 Depends(get_session) 給的 session 就關閉了。如果你在背景任務裡直接用那個 session，會出現 SQLAlchemy: This Session's transaction has been rolled back 錯誤。對應處理：背景任務要自己額外開啟 session，避免踩到已關閉的交易邊界；像 _send_booking_confirmation 那樣寫 Session(engine) 重連一次，這樣即使 request 結束，背景任務仍能用一份全新的 session 讀寫資料。

第二個常見踩雷是「APScheduler 預設 timezone 不對」。如果你不在 AsyncIOScheduler(timezone="UTC") 明確指定，3.x 預設會用機器的當地時區，部署到雲端主機（通常是 UTC）就會發生「早上九點」被解讀成 UTC 早上九點，與使用者預期的「當地早上九點」差八小時。對應處理：永遠明確傳 timezone="UTC"，並讓寄送內容的時間戳記也用 UTC，UI 端再換算當地時區。

第三個錯誤是「測試裡 apscheduler 真的在背景觸發」。如果你直接在 test_create_booking_creates_queued_notification 裡期待寄送在 200 毫秒內完成，可能會因為事件迴圈時序問題而 flaky。對應處理：測試時把 scheduler 整個關掉（scheduler.shutdown(wait=False)）或注入 Fake 排程器；驗 queued 通知存在即可，不必真等寄送完。

第四個是「queued 通知的 body 太長塞不下欄位」。SQLite 對 TEXT 欄位幾乎沒限制，但 PostgreSQL 把 Notification.body 設成 VARCHAR(2000) 就會炸。對應處理：把 body 寫成 Text，並在 application 層用 Pydantic 限制最大長度（1000–2000 字元）。

效能與實務提醒

BackgroundTasks 的本質是「在同一個 process 的事件迴圈裡執行」，所以 CPU bound 的工作（例如編碼 PNG、壓縮影片）不該放 BackgroundTasks；那種該用 Celery 5.4 或 Arq。本專案的「模擬寄信」只是寫 log，所以 BackgroundTasks 完全夠用。要注意：BackgroundTasks 不會自動重試，失敗要自己補機制；我們剛好把重試邏輯寫在 deliver_notification() 裡並寫進 AuditLog，是另一種「保證有追蹤、不保證一次成功」的設計。

APScheduler 3.10 在單機 process 內執行沒有問題，但若未來要水平擴展（多個 API process 同時跑），每台機器都會各自送一份提醒，客戶就會收到好幾封。對應處理：水平擴展時把排程工作搬到一台專屬的 worker（用 Celery beat、Arq cron、k8s CronJob 都行），API process 不再啟動 scheduler。今天我們只有單機，所以 lifespan 直接啟動是對的；未來升級時再重構。

另一個實務提醒是 AuditLog 會迅速膨脹。一家中型預約系統一天可能累積 5,000 筆 audit 紀錄（booking.queued、notification.sent、booking.cancelled 等），半年就是 90 萬筆。對應處理：定期把超過 180 天的 AuditLog 搬到冷儲存（NAS、BigQuery、ClickHouse），保留近半年供即時查詢。今天我們先寫進同一個資料表；Day 42 會展示怎麼用 partition table 或週期性歸檔把這件事做好。

最後，模擬通知的 sink 設計有一個好處：測試時可以注入 sink: list[dict] 驗證內容，production 時換成 logger，不必改業務邏輯。這種「介面固定、實作可換」的寫法，是讓程式能在不同階段（開發、測試、staging、生產）共享同一條呼叫路徑的關鍵。當 production 要切到 SMTP 時，只要新增 SMTPEmailSender，在環境變數 BOOKING_EMAIL_PROVIDER=smtp 時被選用即可，背景任務、路由、模型都不用改。

小結

今天把通知與背景任務兩條線都補進預約管理系統。我們在 Day 35 的模型家族裡新增了 Notification（含 channel、status、attempt_count）與 AuditLog；寫了 SimulatedEmailSender 與 queue_notification()／deliver_notification() 兩階段介面；用 FastAPI 的 BackgroundTasks 把 POST /bookings/ 的確認信切到非同步；用 APScheduler 3.10 註冊了「明天 09:00 提醒」與「每天 03:00 清 7 天前的 QUEUED」兩個 cron 工作，並整合到 FastAPI 0.116 的 lifespan。整套通知流程 CPU 跑得動，模擬通道不串接真實服務，所有事件都會寫進 AuditLog 方便日後查驗。

結語

今天的重點是把「使用者期待」變成「系統行為」：客戶按下預約那一刻，系統就同步寫好 queued 通知、用 BackgroundTasks 在背景把信寄出（模擬）、並用排程器在固定時間再提醒與清理。我們刻意把通知通道寫成可替換介面，讓模擬、SMTP、簡訊商可以在不同階段無痛切換。明天，我們會用這些基礎建設進入「後台介面（HTMX）」的世界：管理者登入後能在瀏覽器上看到今日預約、手動發送通知、批次改期，整個頁面不用寫一行 SPA／React 程式碼，只要 HTMX 的 hx-get、hx-post 與伺服器回傳的 HTML 片段就能完成。

延伸資源

  • FastAPI BackgroundTasks 官方說明（0.116，2025）：https://fastapi.tiangolo.com/tutorial/background-tasks/，定義、限制與注意事項。
  • APScheduler 官方文件（3.10，2024-2025）：https://apscheduler.readthedocs.io/en/3.x/，AsyncIOScheduler、CronTrigger 與錯誤處理模式。
  • SQLModel Session 與交易邊界（0.0.24，2025）：https://sqlmodel.tiangolo.com/advanced/session/，說明 background task 內為什麼要自己開 session。
  • Python asyncio 任務與 APScheduler 的整合範例（2025）：https://docs.astral.sh/uv/，uv 管理 apscheduler==3.10.* 與 sqlmodel==0.0.24 的寫法。
  • httpx ASGITransport（0.28，2025）：https://www.python-httpx.org/async/#calling-into-python-web-apps，在 pytest 內呼叫 ASGI app 的標準寫法。
標籤：

留言

張貼留言

這個網誌中的熱門文章

Day 2 變數與資料型別

Day 2 變數與資料型別 引言 寫程式的過程中，變數與資料型別是處理資料的基礎。變數是存放資料的容器，資料型別則決定這筆資料有哪些特性、可以進行哪些操作。學會定義變數、認識各種資料型別，是學好 Python 的關鍵一步。 這篇文章會帶你了解 Python 中變數的觀念、如何定義變數，以及常見的資料型別，包括整數、浮點數、字串、布林值，還有串列、元組、字典與集合等容器型別。我們也會介紹變數的命名規則與撰寫風格建議，以及如何用 type() 檢查資料型別。 什麼是變數？如何在 Python 中定義變數 變數是在程式執行時用來存放資料的名稱。透過定義變數，我們可以給一筆資料一個名字，並在程式的其他地方用這個名字取用該筆資料。在 Python 中，變數不需要事先宣告型別，因為 Python 是動態型別語言，變數的型別由指定給它的值決定。 定義變數的基本語法 在 Python 中定義變數非常簡單，只要用賦值符號 = 把值指定給變數即可。例如： x = 5 # 定義變數 x，並把整數 5 賦值給它 name = "Alice" # 定義變數 name，並把字串 "Alice" 賦值給它 在這裡，x 是一個變數，被賦予整數 5；name 是另一個變數，被賦予字串 "Alice"。 變數的更新與覆寫 變數的值可以修改，也就是說，我們可以在程式的不同地方給同一個變數新的值。例如： x = 10 # x 最初被賦予 10 x = 15 # x 的值現在被更新為 15 這樣就能依照需求，在程式執行過程中靈活調整變數的值。 Python 的動態型別系統 Python 和某些靜態型別語言不同，定義變數時不需要宣告型別。賦值時，Python 會根據值自動判斷變數的型別。例如： x = 5 # x 是整數 x = 3.14 # x 變成浮點數 x = "Hi" # x 變成字串 同一個變數在程式執行過程中可以存放不同型別的值，這是 Python 的彈性之一。 常見資料型別 在 Python 中，資料型別決定我們可以對變數進行哪些操作...
張貼留言
閱讀完整內容

Day 1 Python 簡介與環境設定

Day 1 Python 簡介與環境設定 引言 在現在的科技環境裡，程式設計已經是一項重要技能。無論你是對資料科學有興趣、想成為開發者，或是想踏入人工智慧（AI）領域，學會寫程式都能明顯提升你的競爭力。在眾多程式語言中，Python 因為語法簡單、功能強大、應用範圍廣泛，成為許多人進入程式世界的第一選擇。這篇文章會帶你認識 Python 的背景與優勢，並一步步教你在不同系統上安裝與設定 Python 開發環境，最後寫出第一支 Python 程式。 為什麼選擇 Python？ Python 是一種高階程式語言，由 Guido van Rossum 在 1991 年發布。Python 的設計哲學強調程式碼的可讀性，並用縮排來定義程式區塊，這點和許多使用大括號的語言不同。簡潔的語法讓它成為初學者的理想選擇；就算是經驗豐富的開發者，也能用它完成複雜的專案。 Python 的優勢如下： 簡單易學 ：Python 的語法清楚、結構簡潔，初學者很快就能上手。和其他語言相比，學習曲線相對平緩，不需要先弄懂一堆複雜觀念，就能開始寫程式。 應用範圍廣泛 ：從資料科學、網頁開發、人工智慧、機器學習、自動化測試到網路爬蟲，Python 都有大量開源函式庫與工具支援，而且在這些領域都扮演關鍵角色。 豐富的函式庫與框架 ：Python 的函式庫生態系非常龐大。做資料分析有 NumPy、Pandas；開發網站有 Django、Flask；做深度學習有 TensorFlow、PyTorch。各種需求幾乎都能找到對應的套件，讓開發更有效率。 跨平台支援 ：Python 支援 Windows、macOS、Linux 等作業系統，程式通常不需要太多修改就能跨平台執行，讓開發與部署更有彈性。 活躍的社群 ：Python 擁有龐大的開發者社群。學習或開發上遇到問題，幾乎都能在社群與論壇（例如 Stack Overflow）找到答案，對初學者來說是很強的後盾，也能減少卡關時的挫折感。 Python 的應用領域 Python 的流行與強大功能，讓許多領域都開始大量使用它。以下是幾個常見的應用方向： 資料科學 ：隨著大數據與人工智慧興起，資料科學大量使用 Python。NumPy、Pandas 與 Matplotlib 等工具能處理和分析龐...
張貼留言
閱讀完整內容

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門

Python 從入門到 PyTorch 深度學習：開啟 AI 世界的大門 隨著人工智慧（AI）與深度學習（Deep Learning）快速發展，越來越多人對這些技術產生興趣。不論你是想踏入 AI 領域的初學者，還是已經有程式基礎的開發者，學好 Python 與深度學習框架（例如 PyTorch），都能為你打開更多可能。 為什麼選擇 Python？ Python 已經是資料科學與人工智慧領域的首選語言。它的語法簡潔、容易上手，而且擁有龐大的生態系與大量開源函式庫。無論是資料處理、資料視覺化，還是建立機器學習與深度學習模型，Python 都能勝任。對想進入 AI 或資料科學領域的人來說，它幾乎是必備工具。 PyTorch 是什麼？ PyTorch 是由 Meta（原 Facebook）AI 研究團隊開發的開源深度學習框架，以易用、靈活和動態計算圖著稱，是許多 AI 研究人員與開發者的首選。相較於其他框架，PyTorch 的寫法更貼近原生 Python，對初學者相對友善。無論是簡單的實驗，還是複雜的深度學習模型，PyTorch 都能提供強大的支援。 這個系列能帶給你什麼？ 這個系列會從 Python 的基礎開始，帶你一步一步學習，最後能自己用 PyTorch 建立深度學習模型。即使你完全沒有寫過程式，也能跟著文章的節奏累積技能，理解 AI 與深度學習的核心觀念。 本系列涵蓋的主題 Python 基礎：從變數、條件判斷到函式與模組。 資料處理工具：用 NumPy 與 Pandas 有效率地操作資料。 資料視覺化：用 Matplotlib 與 Seaborn 把資料畫成圖表。 深度學習的數學基礎：線性代數、微積分與機率。 PyTorch 入門：理解張量、模型建構與 GPU 加速。 基礎深度學習模型：CNN 與 RNN 的實作應用。 深度學習專案實戰：從資料前處理到模型部署的端到端流程。 誰適合這個系列？ 程式初學者 ：如果你對 AI 充滿好奇，卻還沒寫過程式，系列的第一部分會帶你快速上手 Python，並幫助你理解深度學習的基本觀念。 資料科學愛好者 ：如果你已經熟悉一些資料處理方法，進階部分會教你如何用 PyTorch 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容