跳到主要內容

Web Day 37 專案：預約流程與衝突檢查



Web Day 37 專案：預約流程與衝突檢查

執行需求：CPU 可跑。今天是預約管理系統的核心業務邏輯：客戶建立預約、查可預約時段、取消預約。我們會用具時間區間重疊演算法（interval overlap）確保「同一個服務在重疊時段不會被兩個 customer 預約到」，並用 transaction 隔離級別與 SELECT FOR UPDATE 確保 race condition 下也只有一個能成功。今天所有資料都是虛構的預約管理系統範例，服務名稱、客戶姓名、時段都不是真實的。

引言

預約管理系統的核心挑戰是「同一時段不能被兩個人預約」。聽起來簡單，但實作起來會碰到三個層次的難題：第一，純應用層的「先查現有 booking 再決定能不能下單」在兩個使用者同時按按鈕時會失效；第二，service 的 buffer time（準備時間）讓重疊判定變得複雜——「同一時段」不只是「同一秒」，而是要預留服務前後的準備時間；第三，時區轉換讓「同一個台灣時間下午三點」在不同時區的使用者看起來都不一樣，但 DB 裡存的是同一個 UTC 時間。

今天會把這三層難題一次解決。我們會寫一個 conflict check 函式：給定 (start_at, end_at, service_id)，回傳是否有衝突；寫一個 create booking 的 endpoint，用 transaction 包住「查衝突 → 建 booking」，用 SELECT FOR UPDATE 鎖住相關 row 避免 race condition；再寫一個 list_available_slots 的 endpoint，依 TimeSlot 展開成可預約的具體時間。所有程式碼都會在 Day 35 的共用設定上擴充，不改既有 entity 結構。

從前面兩天沿用什麼

今天的程式碼完全建立在 Day 35（資料模型）與 Day 36（認證）的基礎上，引用以下東西：

  • Booking、Service、TimeSlot、BookingStatusLog model
  • BookingStatus enum（pending / confirmed / cancelled / completed）
  • session_scope context manager（transaction 邊界）
  • require_customer 與 require_admin dependency（角色限制）
  • create_access_token 與 current_user 認證鏈

如果你還沒跑過 `alembic upgrade head` 與 `python scripts/seed.py`，請先執行。seed 會建立 3 個虛構服務、10 個 customer、20 筆 booking，這些資料會被今天的範例 endpoint 直接用到。

區間重疊演算法

兩個時間區間 [a_start, a_end) 與 [b_start, b_end) 重疊的數學條件是「a_start 小於 b_end 且 b_start 小於 a_end」。這條公式看起來很簡單，但要記得幾個關鍵細節：第一，區間是「半開」的—— [14:00, 15:00) 與 [15:00, 16:00) 不算重疊，因為前者 15:00 結束、後者 15:00 開始，剛好相接；第二，「等於」不算重疊（a_start == b_end 不算）；第三，時段對齊時要特別小心 buffer time——兩個服務中間要留 15 分鐘準備，這 15 分鐘要算進重疊判定。順帶一提，「半開區間」的選擇不是品味問題：如果改用閉區間判定，相鄰的兩個時段會在交界點「重疊」，導致永遠無法相鄰預約，這是時間系統設計最常見的第一個坑。

把 buffer 也算進去的公式變成：先把 b 的 [b_start, b_end) 往外擴張 buffer_minutes，再判斷與 a 是否重疊。或者寫成 SQL：找出任何 booking 的 (start_at - buffer_minutes) 落在新預約的 [start_at, end_at) 範圍內、或新預約的 (end_at + buffer_minutes) 落在既有 booking 的範圍內、或兩者大幅重疊。我們這裡採第一種做法，在 Python 端做重疊判定。

"""app/booking_conflict.py：預約衝突檢查邏輯（純函式、無 DB 依賴）。"""
from dataclasses import dataclass
from datetime import datetime, timedelta


@dataclass(frozen=True)
class TimeRange:
    start_at: datetime
    end_at: datetime

    def overlaps(self, other: "TimeRange", buffer: timedelta = timedelta(0)) -> bool:
        """兩個時間區間是否重疊（含 buffer）。"""
        expanded_self = TimeRange(self.start_at, self.end_at + buffer)
        expanded_other = TimeRange(other.start_at - buffer, other.end_at)
        return expanded_self.start_at < expanded_other.end_at and expanded_other.start_at < expanded_self.end_at


def find_conflict(target: TimeRange, existing: list[TimeRange], buffer: timedelta) -> TimeRange | None:
    """從既有 booking 找出第一個與 target 重疊的區間，沒有就回 None。"""
    for r in existing:
        if target.overlaps(r, buffer=buffer):
            return r
    return None

這份程式碼把重疊判定封裝在 TimeRange dataclass，方便單獨測試。`overlaps` 函式的條件是「self.start < other.end AND other.start < self.end」，這是區間重疊的標準寫法。`buffer` 參數讓重疊判定可以往外擴張，模擬 buffer time 的影響。注意這裡用「半開」區間——end 是 exclusive——所以「14:00-15:00」與「15:00-16:00」不算重疊，這對排程系統來說是正確的語意。

為了驗證這個演算法，我們寫一組 pytest：

"""tests/test_booking_conflict.py：區間重疊演算法的單元測試。"""
from datetime import datetime, timedelta

from app.booking_conflict import TimeRange, find_conflict


def _r(start_h: int, end_h: int) -> TimeRange:
    base = datetime(2025, 7, 20, 0, 0, 0)
    return TimeRange(base + timedelta(hours=start_h), base + timedelta(hours=end_h))


def test_adjacent_ranges_do_not_overlap():
    assert not _r(14, 15).overlaps(_r(15, 16))


def test_overlapping_ranges_overlap():
    assert _r(14, 16).overlaps(_r(15, 17))


def test_contained_range_overlaps():
    assert _r(14, 18).overlaps(_r(15, 16))


def test_disjoint_ranges_do_not_overlap():
    assert not _r(9, 10).overlaps(_r(14, 16))


def test_buffer_expands_range():
    target = _r(15, 16)
    existing = _r(14, 15)
    assert not target.overlaps(existing)
    assert target.overlaps(existing, buffer=timedelta(minutes=15))


def test_find_conflict_returns_first():
    existing = [_r(9, 10), _r(15, 16), _r(16, 17)]
    conflict = find_conflict(_r(15, 30, 16, 30) if False else _r(15, 30, 16, 30), existing, timedelta())
    assert conflict is not None
    assert conflict.start_at.hour == 15

這組測試覆蓋了五個關鍵情境：相接不算重疊、明顯重疊、包含關係、毫無交集、buffer 影響。`overlaps` 是純函式，跑得快也容易理解，可以放心放進正式程式碼。`find_conflict` 雖然是 O(n)，但實務上 n 通常是當天的 booking 數（幾筆到幾十筆），效能不是問題；如果未來需要更快，可以改用 SQL 端的 EXISTS 子查詢配合 Day 35 加的 start_at / end_at index。

在 SQLAlchemy 端做 conflict 查詢

純函式版適合做「給定清單」的判定，但實際查 booking 還是要從 DB 撈。我們寫一個函式，把「找出重疊 booking」這件事直接做成 SQLAlchemy query，搭配 SELECT FOR UPDATE 在 transaction 內鎖住結果集，避免 race condition。

"""app/repositories/bookings.py：booking 的查詢與寫入邏輯。"""
from datetime import datetime, timedelta
from uuid import uuid4

from sqlalchemy import and_, select
from sqlalchemy.orm import Session

from app.models import Booking, BookingStatus, BookingStatusLog, Service


def fetch_active_bookings(
    session: Session,
    service_id: str,
    range_start: datetime,
    range_end: datetime,
    with_for_update: bool = False,
) -> list[Booking]:
    """撈出某服務在指定時段內的 active booking（含 pending 與 confirmed）。"""
    stmt = (
        select(Booking)
        .where(
            Booking.service_id == service_id,
            Booking.status.in_([BookingStatus.pending, BookingStatus.confirmed]),
            Booking.deleted_at.is_(None),
            Booking.start_at < range_end,
            Booking.end_at > range_start,
        )
        .order_by(Booking.start_at)
    )
    if with_for_update:
        stmt = stmt.with_for_update()
    return list(session.execute(stmt).scalars())


def create_booking(
    session: Session,
    *,
    customer_id: str,
    service: Service,
    start_at: datetime,
    notes: str = "",
) -> Booking:
    """建立一筆 booking（含 conflict 檢查與 audit log）。"""
    end_at = start_at + timedelta(minutes=service.duration_minutes)
    buffer = timedelta(minutes=service.buffer_minutes)

    existing = fetch_active_bookings(
        session,
        service.id,
        start_at - buffer,
        end_at + buffer,
        with_for_update=True,
    )

    target_range = __import__("app.booking_conflict", fromlist=["TimeRange"]).TimeRange(start_at, end_at)
    for b in existing:
        candidate = __import__("app.booking_conflict", fromlist=["TimeRange"]).TimeRange(b.start_at, b.end_at)
        if target_range.overlaps(candidate, buffer=buffer):
            raise ConflictError(f"時段 {b.start_at.isoformat()} 已被預約")

    reference = f"BS-{start_at.strftime('%Y%m%d')}-{uuid4().hex[:6].upper()}"
    booking = Booking(
        reference=reference,
        customer_id=customer_id,
        service_id=service.id,
        start_at=start_at,
        end_at=end_at,
        status=BookingStatus.pending,
        notes=notes,
    )
    session.add(booking)
    session.flush()

    log = BookingStatusLog(
        booking_id=booking.id,
        from_status=None,
        to_status=BookingStatus.pending,
        changed_by=customer_id,
        reason="建立預約",
    )
    session.add(log)
    return booking


class ConflictError(Exception):
    """預約時段衝突。"""

這份 repository 有四個關鍵設計：第一，`fetch_active_bookings` 用 `start_at < range_end AND end_at > range_start` 這組條件做索引掃描，這正是 Day 35 加 start_at / end_at index 的目的；第二，`with_for_update=True` 加上 transaction 等於「SELECT FOR UPDATE」，在 Postgres 會對匹配的 row 加 row-level lock，避免其他 transaction 同時修改；第三，conflict 檢查在 Python 端跑，這對 buffer 邏輯更直覺；第四，建立 booking 的同時寫一筆 BookingStatusLog，這樣審計紀錄與 booking 同步存在。

ConflictError 是自訂例外，FastAPI 的 exception handler 可以把它轉成 409 Conflict。Day 38 通知章節會再示範怎麼把例外轉成對客服友善的訊息。

用 Postgres 的 EXCLUDE 條件約束當最後防線

Python 端的衝突檢查是「應用層防線」，還有一層更硬的保險可以放在資料庫層：PostgreSQL 的 EXCLUDE 條件約束（搭配 btree_gist 擴充套件與 tstzrange 型別）。它的意思是「同一個 service_id 底下，任何兩筆 active booking 的時間區間不允許重疊」，資料庫會在 INSERT 違反時直接回 40P01 錯誤。這條防線的價值在於：就算未來有人寫了新的程式路徑（例如匯入腳本、管理後台直連 DB），繞過了 Python 的檢查，資料庫仍然會擋下重疊的預約。

# Alembic 遷移中加入 EXCLUDE 條件約束（Postgres 17）

from alembic import op

def upgrade():
    op.execute("CREATE EXTENSION IF NOT EXISTS btree_gist")
    op.execute("""
        ALTER TABLE bookings
        ADD CONSTRAINT no_overlapping_bookings
        EXCLUDE USING gist (
            service_id WITH =,
            tstzrange(start_at, end_at) WITH &&
        ) WHERE (deleted_at IS NULL AND status != 'cancelled')
    """)

def downgrade():
    op.execute("ALTER TABLE bookings DROP CONSTRAINT no_overlapping_bookings")

要注意的是 EXCLUDE 約束只在 PostgreSQL 有效（SQLite 沒有這個功能），而且它的重疊判定不包含 buffer——如果你需要 15 分鐘緩衝，可以把範圍寫成 tstzrange(start_at - interval '15 minutes', end_at + interval '15 minutes') 之類的變形。應用層檢查與資料庫約束各司其職：前者給使用者友善的錯誤訊息，後者保證資料在極端情況下依然一致。

create booking endpoint

把 repository 接上 FastAPI endpoint，並用 Day 36 的 require_customer dependency 限制只有 customer 能下單。我們也用 Pydantic 做 request/response schema 驗證。

"""app/routers/bookings.py：booking 的 endpoint。"""
from datetime import datetime

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from app.deps import get_session, require_customer
from app.models import Booking, BookingStatus, Service, User
from app.repositories.bookings import ConflictError, create_booking, fetch_active_bookings


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


class CreateBookingRequest(BaseModel):
    service_id: str
    start_at: datetime
    notes: str = Field(default="", max_length=500)


class BookingResponse(BaseModel):
    id: str
    reference: str
    service_id: str
    start_at: datetime
    end_at: datetime
    status: BookingStatus


@router.post("/", response_model=BookingResponse, status_code=status.HTTP_201_CREATED)
def create_booking_endpoint(
    payload: CreateBookingRequest,
    session: Session = Depends(get_session),
    user: User = Depends(require_customer),
) -> BookingResponse:
    service = session.get(Service, payload.service_id)
    if service is None or not service.is_active or service.deleted_at is not None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="服務不存在或已停用")

    try:
        booking = create_booking(
            session,
            customer_id=user.id,
            service=service,
            start_at=payload.start_at,
            notes=payload.notes,
        )
        session.flush()
    except ConflictError as exc:
        raise HTTPException(status.HTTP_409_CONFLICT, detail=str(exc))

    return BookingResponse(
        id=booking.id,
        reference=booking.reference,
        service_id=booking.service_id,
        start_at=booking.start_at,
        end_at=booking.end_at,
        status=booking.status,
    )

這個 endpoint 把 Day 35 的 service 查詢、Day 36 的 customer 認證、今天的 conflict 檢查與寫入全部串起來。`require_customer` 確保只有 customer 角色能呼叫（admin 想要下單測試可以用 admin token，但業務邏輯上 admin 通常是用後台建立測試訂單，Day 39 會實作這部分）。`session.flush()` 把 booking 與 log 寫進 session，但還沒 commit，要靠 session_scope 的 context manager 結束時才 commit，這樣 conflict 時整個 transaction 會 rollback、不會留下半套 booking。

取消 booking

取消流程比建立簡單：撈出 booking、檢查是不是當事人的、檢查是否已過了開始時間、把 status 改成 cancelled 並寫 audit log。這個 endpoint 開放給 booking 主人或 admin 呼叫。

@router.post("/{booking_id}/cancel", response_model=BookingResponse)
def cancel_booking(
    booking_id: str,
    session: Session = Depends(get_session),
    user: User = Depends(get_current_user),
) -> BookingResponse:
    booking = session.get(Booking, booking_id)
    if booking is None or booking.deleted_at is not None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="booking 不存在")

    is_owner = booking.customer_id == user.id
    is_admin = user.role.value == "admin"
    if not (is_owner or is_admin):
        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="只能取消自己的預約")

    if booking.status in (BookingStatus.cancelled, BookingStatus.completed):
        raise HTTPException(status.HTTP_409_CONFLICT, detail="booking 已取消或已完成")

    if booking.start_at < datetime.utcnow() and not is_admin:
        raise HTTPException(status.HTTP_409_CONFLICT, detail="已過開始時間，無法取消")

    previous = booking.status
    booking.status = BookingStatus.cancelled
    booking.cancelled_at = datetime.utcnow()
    session.add(BookingStatusLog(
        booking_id=booking.id,
        from_status=previous,
        to_status=BookingStatus.cancelled,
        changed_by=user.id,
        reason="取消預約",
    ))

    return BookingResponse(
        id=booking.id,
        reference=booking.reference,
        service_id=booking.service_id,
        start_at=booking.start_at,
        end_at=booking.end_at,
        status=booking.status,
    )

取消流程展示了「多角色共享 endpoint」的標準做法：`get_current_user`（不限角色）取得使用者，再用 `is_owner or is_admin` 在 endpoint 內判斷權限。這跟 Day 36 的 require_admin/require_customer 互補：前者是「必須是某個角色」，後者是「特定 owner 或 admin」。對 side project 來說這層邏輯夠用；對正式環境要再細分（例如 staff 可以取消但不能建立），可以把 `is_admin` 改成 role-based 的 helper。

列出可預約時段

客戶在建立預約之前要能看到「哪些時段還能選」。我們用 Day 35 的 TimeSlot 模板展開成具體時間、再過濾掉已被預約的時段，回傳給前端。

"""app/services/availability.py：展開 TimeSlot 並過濾已預約時段。"""
from datetime import datetime, time, timedelta

from sqlalchemy.orm import Session

from app.models import BookingStatus, Service, TimeSlot
from app.repositories.bookings import fetch_active_bookings


def list_available_slots(
    session: Session,
    service: Service,
    from_date: datetime,
    to_date: datetime,
    step_minutes: int = 30,
) -> list[datetime]:
    """展開某服務在 [from_date, to_date) 範圍內的所有可預約起點。"""
    slots: list[datetime] = []
    cursor = from_date
    while cursor < to_date:
        weekday = cursor.weekday()
        matching = session.query(TimeSlot).filter(
            TimeSlot.service_id == service.id,
            TimeSlot.day_of_week == weekday,
            TimeSlot.is_active.is_(True),
        ).all()

        for ts in matching:
            start_h, start_m = map(int, ts.start_time.split(":"))
            end_h, end_m = map(int, ts.end_time.split(":"))
            slot_start = cursor.replace(hour=start_h, minute=start_m, second=0, microsecond=0)
            slot_end = cursor.replace(hour=end_h, minute=end_m, second=0, microsecond=0)
            while slot_start + timedelta(minutes=service.duration_minutes) <= slot_end:
                slots.append(slot_start)
                slot_start += timedelta(minutes=step_minutes)

        cursor += timedelta(days=1)
        cursor = cursor.replace(hour=0, minute=0, second=0, microsecond=0)

    slots = sorted(set(slots))

    if not slots:
        return []

    range_start = slots[0]
    range_end = slots[-1] + timedelta(minutes=service.duration_minutes)
    bookings = fetch_active_bookings(session, service.id, range_start, range_end)
    booked_starts = {b.start_at for b in bookings}

    return [s for s in slots if s not in booked_starts]

這份展開邏輯有兩個關鍵設計：第一，先展開再過濾——把所有候選時間列出來，再用 fetch_active_bookings 一次撈出既有 booking 過濾掉，這比「每個候選時間都查一次 DB」少很多次 query；第二，用 `set` 與 `in` 操作確保展開不重複，並用 sorted 確保回傳結果是時序的。

展開的 step_minutes 預設 30 分鐘，這對 30 分鐘以下的服務剛好；對 60 分鐘的服務可以拉長到 60 分鐘（避免每 30 分鐘都有候選）。step 是「時間顆粒度」的概念，跟 service duration 無關：可以讓 60 分鐘的服務每 30 分鐘一個候選（彈性高），也可以每 60 分鐘（強制整點）。今天先用 30 分鐘當範例，後續可以改成 service 欄位。

測試預約流程的關鍵情境

預約流程最容易出 bug 的是 race condition 與 edge case。我們寫一組整合測試覆蓋「正常下單」、「衝突下單」、「取消後空出時段」、「admin 強制取消」四個關鍵路徑。

"""tests/test_bookings.py：booking endpoint 的整合測試。"""
import os
import uuid

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import select

from app.db import session_scope
from app.main import app
from app.models import Booking, BookingStatus, Service, User, UserRole
from app.security import hash_password


@pytest.fixture(autouse=True)
def env(monkeypatch, tmp_path):
    db = tmp_path / "test.db"
    monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db}")
    monkeypatch.setenv("JWT_SECRET", "test-secret-32-bytes-long-aaaaaaaaaa")


@pytest.fixture
def client():
    return TestClient(app)


@pytest.fixture
def seeded():
    with session_scope() as session:
        customer = User(
            id=str(uuid.uuid4()),
            email="c@example.com",
            password_hash=hash_password("pw-12345678"),
            display_name="customer",
            role=UserRole.customer,
        )
        service = Service(name="test", duration_minutes=60, buffer_minutes=15, price_cents=1000)
        session.add(customer)
        session.add(service)
        session.flush()
        return {"customer_id": customer.id, "service_id": service.id}


def test_create_booking_success(client, seeded):
    token = client.post("/auth/login", json={"email": "c@example.com", "password": "pw-12345678"}).json()["access_token"]
    res = client.post(
        "/bookings/",
        headers={"Authorization": f"Bearer {token}"},
        json={"service_id": seeded["service_id"], "start_at": "2025-07-20T15:00:00+00:00"},
    )
    assert res.status_code == 201
    assert res.json()["status"] == "pending"


def test_create_booking_conflict(client, seeded):
    token = client.post("/auth/login", json={"email": "c@example.com", "password": "pw-12345678"}).json()["access_token"]
    headers = {"Authorization": f"Bearer {token}"}
    body = {"service_id": seeded["service_id"], "start_at": "2025-07-20T15:00:00+00:00"}
    client.post("/bookings/", headers=headers, json=body)
    res = client.post("/bookings/", headers=headers, json=body)
    assert res.status_code == 409

這組測試覆蓋了「下單成功」與「下單衝突」兩個關鍵路徑。SQLite 的 transaction 隔離行為跟 Postgres 略有不同（SQLite 用 BEGIN IMMEDIATE），但 conflict 檢查的 SQL 條件是相同的，所以測試在兩個環境都有效。對 race condition 的真正測試需要多執行緒模擬，今天先用單執行緒版本驗證邏輯正確性，並在 Day 43 壓測章節用 locust 跑並發測試。

沒有 Docker 時的替代流程

今天的程式碼完全可以在本機用 SQLite + venv 跑：

export DATABASE_URL="sqlite:///./booking.db"
export JWT_SECRET="$(openssl rand -hex 32)"
alembic upgrade head
python scripts/seed.py
ADMIN_PASSWORD="my-strong-pw" python scripts/set_admin_password.py
uvicorn app.main:app --reload

打開 http://127.0.0.1:8000/docs 用 customer token 呼叫 POST /bookings/，故意帶兩個相同時段可以驗證 409 衝突。SQLite 對 SELECT FOR UPDATE 的支援有限（會忽略 lock），所以 SQLite 模式下的 race condition 防護不完整；要真正驗證並發行為還是要用 Day 31 的 Postgres docker compose。

常見錯誤與踩雷

忘記把 buffer 算進重疊判定。最容易寫錯的版本是直接用 `start_at < existing.end_at AND existing.start_at < end_at` 比對，沒有加 buffer。這會讓「14:00 結束的服務」與「14:00 開始的下一個服務」都能成功預約，但實務上兩個服務中間沒有 15 分鐘緩衝會出問題。我們的 TimeRange.overlaps 有 buffer 參數，repository 端也傳了 service.buffer_minutes，兩層保護避免漏寫。

用 datetime.now() 比較而不用 UTC。「已過開始時間」的檢查如果用 `datetime.now()` 會拿到本地時間，跟 DB 裡的 UTC 時間差 8 小時，會誤判某些時段為「已過」或「未過」。務必用 `datetime.now(timezone.utc)` 跟 DB 的 TIMESTAMP WITH TIME ZONE 比較。

在 endpoint 內寫 query 但忘了 transaction。我們的 endpoint 用 session_scope() 包住整個 request，conflict check 與 insert 都在同一個 transaction 內。如果忘了包、或用新的 session 開 transaction，conflict check 與 insert 之間會被其他 transaction 插入 booking，造成 race condition。Day 31 的 session_scope 設計就是為了解決這個問題，務必沿用。

取消時沒寫 audit log。狀態變更一定要寫 BookingStatusLog，否則客服遇到爭議時無法追溯。我們的 create_booking 與 cancel_booking 都有寫 log，後續 Day 38 通知章節會用這些 log 觸發通知。

buffer 的單位搞錯。TimeRange.overlaps 的 buffer 參數是 timedelta，如果呼叫端誤把「15」當成 timedelta 傳入（想的是 15 分鐘），Python 會在運算時直接拋 TypeError；另一種更隱密的錯法是把 15 分鐘想成 15 小時——用 timedelta(hours=15) 而非 timedelta(minutes=15)，重疊判定會變得極端保守，幾乎所有相鄰時段都被擋。對應排查方向：在 repository 端把最終的 buffer 值印出來檢查，並在 Service 模型層用 Pydantic 的欄位驗證限制 buffer_minutes 的合理範圍（例如 0 到 60）。

UTC 邊界的日切問題。「今天的預約」在 UTC 換日時會出現落差：台灣時間 9 月 1 日凌晨一點，在 UTC 是 8 月 31 日 16:00。如果 fetch_active_bookings 的日期範圍由伺服器本地日期計算，跨時區部署時會撈錯天的資料。務必讓「天」的邊界由業務時區（例如 Asia/Taipei）決定、轉成 UTC 存入查詢，而不是用 datetime.utcnow() 的日期直接切。

列出可預約時段忘記過濾已過時間。如果使用者在中午打開預約頁，展開的時段包含早上 9 點，建立時才被擋下「已過開始時間」，體驗很差。正確做法是在展開迴圈裡加一行過濾：slot_start 加上服務時長後仍晚於「現在 + 最短提前量」才保留。最短提前量（例如 2 小時前不可預約）應該做成 Service 的欄位而不是寫死在程式裡，這樣不同服務可以有不同的提前量規則。

效能與實務提醒

區間重疊演算法本身是 O(n) 對單一服務的當天 booking，n 通常小於 100，效能不是問題。但如果你的服務量很大、想要進一步加速，可以把 fetch_active_bookings 的範圍縮小——從「整天」縮小到「候選時段前後各 buffer」就夠了。我們目前的實作已經這麼做，但實務上可以再壓縮，例如只撈「start_at 在 [target - duration, target + duration + buffer] 內」的 booking，能讓 query 計畫用上 start_at index 而不是 range scan。

SELECT FOR UPDATE 在高並發下可能造成 lock contention。對 side project 來說一次只鎖一個服務的 row 沒問題；對大型應用要考慮用 advisory lock 或樂觀鎖（版本號欄位）。今天先用悲觀鎖（SELECT FOR UPDATE）簡化邏輯，未來流量大時再升級。

list_available_slots 在展開大量時段時（例如查整個月）會跑得很慢。實務上建議把查詢範圍限制在 7 天內，超過就分頁。對 side project 來說這個限制可以加在前端：日期選擇器只讓使用者選 7 天內的日期。這層效能優化在 Day 43 壓測章節會再示範。

另一個值得提前設計的是「取消後的名額恢復」。booking 被 cancel 之後，該時段會重新變成可預約——但通知、統計與快取可能還停留在舊狀態。實務上有兩種做法：第一，在 cancel 的同一個 transaction 內把相關快取鍵刪掉（cache invalidation 與資料寫入同進退）；第二，讓快取短生命週期（例如 60 秒 TTL），接受短暫的不一致。今天的範例還沒有快取層，Day 20 的 Redis 快取策略套上來時，記得把「狀態變更 → 快取失效」這條路徑一起設計，否則客戶會看到「其實已被取消的時段還可以預約」的靈異現象。這種跨層的一致性思考，是把「能跑的 API」升級成「能信任的系統」的分水嶺。

小結

今天把預約管理系統的核心業務邏輯實作完成：區間重疊演算法、SELECT FOR UPDATE 防 race condition、建立與取消 endpoint、可預約時段展開。重點觀念包括「半開區間」、buffer time 怎麼影響判定、以及 transaction 隔離怎麼保護並發下單。所有 entity 與共用設定繼續沿用 Day 35 與 Day 36，不破壞既有設計。後續 Day 38 會用 BookingStatusLog 觸發通知、Day 39 的後台介面會用 list_available_slots 顯示月曆、Day 41 一鍵部署時這套邏輯會跟著上線。

結語

預約流程是這個 side project 的核心，今天把「同一時段不能撞」這條規則從觀念落實到程式碼，並加上 transaction 防護與整合測試。明天，我們會把狀態變更串到通知系統：booking 建立或取消時自動發通知給客戶，並用背景任務模擬送出信件。後台的客服也能看到「這筆 booking 為什麼改狀態」的完整軌跡。

延伸資源

  • Postgres SELECT FOR UPDATE 說明：https://www.postgresql.org/docs/17/sql-select.html#SQL-FOR-UPDATE-SHARE
  • SQLAlchemy 2.0 with_for_update：https://docs.sqlalchemy.org/en/20/orm/queryguide/select.html#orm-queryguide-for-update
  • 區間重疊演算法整理：https://en.wikipedia.org/wiki/Interval_tree
  • FastAPI 整合測試（TestClient）：https://fastapi.tiangolo.com/tutorial/testing/
  • passlib argon2 雜湊文件：https://passlib.readthedocs.io/en/stable/lib/passlib.hash.argon2.html
標籤：

留言

張貼留言

這個網誌中的熱門文章

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 建構深度學習模型。 開發者與研究人員 ：想更深入了...
張貼留言
閱讀完整內容