跳到主要內容

Web Day 35 專案定義:預約管理系統的資料模型



Web Day 35 專案定義:預約管理系統的資料模型

執行需求:CPU 可跑。今天開始進入貫穿專案「預約管理系統」(資料全部虛構)。我們要為一家虛構的小型服務業(例如攝影棚、健身教練、諮詢工作室)設計一套線上預約後端,涵蓋服務提供者(管理者)與客戶(使用者)兩種角色、服務品項設定、可預約時段、預約單與衝突檢查。今天專注在資料模型:把 entities、relationships、欄位型別、index 都先決定好,後面 Day 36 認證、Day 37 衝突檢查、Day 39 後台介面、Day 41 一鍵部署都會沿用今天的設計。

引言

前面 34 天我們把 FastAPI、SQLModel、PostgreSQL、Docker、Caddy、CI/CD、健康檢查都學過一遍。今天開始把所有東西組合起來,做一個完整的 side project:預約管理系統。情境設定是「一個人或小團隊提供專業服務、客戶可以線上看時段、預約下單」。這種系統在現實生活中很常見:心理諮商、健身教練、攝影服務、語言家教、清潔服務……業務邏輯簡單,但「同一時段不能被兩個人預約」這條規則讓資料庫設計有幾個關鍵決策要做對。

今天目標是產出一份「後續 10 天的工程都會引用」的資料模型。我們會定義 6 個 SQLModel 類別、寫出對應的 Alembic 遷移、給出一張 ER 圖的純文字描述,並準備好種子資料腳本。所有資料都是虛構的範例——服務名稱、定價、營業時間都不是真實客戶的資料。學完之後你會拿到一個能跑遷移、能用 Faker 灌假資料、能用 pytest 驗證模型一致性的最小骨架。

貫穿專案的共用設定

從 Day 35 到 Day 44,整個預約管理系統的「專案骨架」會保持一致。這裡先把共用設定一次講清楚,避免後續篇章重複定義。

命名空間:所有 SQLModel class 都放在 `app.models` 模組下,從這個模組 import 就能拿到所有 entity。Alembic 設定會讀 `app.models.SQLModel.metadata` 自動產生遷移。

主鍵策略:所有 table 都用 UUID v4 做 primary key(用 Postgres 的 uuid-ossp extension 或 Python 的 uuid4),避免順序 ID 洩漏業務量、也方便未來做 distributed insert。對外顯示用「BS-20250720-0001」這種業務編號是另一層,內部一律用 UUID。

時間欄位:所有 timestamp 用 datetime (timezone aware),DB 端用 TIMESTAMP WITH TIME ZONE,存進去一律 UTC。顯示時再依使用者時區轉換(這個轉換會在後端用 helper function 統一處理,不散落在各個 endpoint)。

軟刪除(soft delete):預約單與客戶資料都不直接 DELETE,而是把 `deleted_at` 設成刪除時間。這樣資料可以回復、稽核 log 可以保留、誤刪可以救援。對應的 query 都會自動加 `WHERE deleted_at IS NULL` 過濾。

密碼欄位:用 `password_hash` 命名(不叫 password),存的是 argon2id 雜湊後的字串。Argon2 是 2025 年 7 月主流的雜湊演算法,passlib 的 CryptContext 提供完整支援。

預設值:所有 enum 欄位都給預設值(例如 BookingStatus.pending),這樣即使前端漏傳欄位也不會壞掉。created_at 與 updated_at 用 server_default 與 onupdate 自動維護。

Entity 總覽

預約管理系統的核心 entity 有六個。我們用一段簡短的描述把每個 entity 與它在系統中的角色講清楚,後面再分別給 SQLModel 定義。

User:系統使用者。區分兩種角色:admin 是服務提供者(管理者),可以管理服務品項、看所有預約、設定可預約時段;customer 是客戶,可以瀏覽服務、建立預約、看自己的歷史。一個帳號同一時間只能有一種角色,但可以升級/降級。User model 同時儲存 timezone 與 display_name,前者用於顯示時區轉換,後者用於對客服訊息與通知。

Service:服務提供者對外提供的服務品項。例如「個人攝影 60 分鐘」、「一對一健身 90 分鐘」、「諮詢 30 分鐘」。每個 service 有 duration(分鐘)、buffer time(準備/收拾時間)、price 與是否啟用。duration 決定 Booking.end_at - start_at 的最小值;buffer 決定兩個 booking 之間要留多少準備時間,這對 Day 37 衝突檢查很重要,因為「時間重疊」要把 buffer 也算進去。

TimeSlot:可預約的時段模板。服務提供者定義「每週三 14:00-17:00」這種週期性時段,系統會自動展開成具體的預約時間。TimeSlot 用 day_of_week + start_time + end_time 表示週期,展開的具體時間存在 Booking 裡。這種設計的好處是「修改一次時段、未來所有展開都會跟著變」,比展開成 datetime 序列再單獨維護簡單很多。

Booking:一次預約單。包含哪個 customer 預約哪個 service、開始時間、結束時間、狀態(pending/confirmed/cancelled/completed)、備註與建立時間。衝突檢查會在這個 table 上做。reference 是對外友善的業務編號(例如 BS-20250720-0001),客服人員拿到這串字號就能查到對應的 booking,比 UUID 容易唸也容易記。

BookingStatusLog:預約狀態變更紀錄。當 booking 從 pending 變成 confirmed(或任何其他轉換)時寫一筆,提供 audit trail。這個 entity 在 Day 38 通知章節會用到。每次狀態轉換都會同時寫 log,這樣客服遇到爭議時可以從 log 反推「是誰在什麼時候改的」。

Notification(Day 38 引入):通知紀錄。模擬寄信結果,不串接真實服務。每筆 booking 狀態變動會建立一筆 Notification,預設 status 是 pending,背景任務模擬送出後改為 sent 或 failed。對 side project 來說這個 entity 已經足夠;如果未來要串接真實的 email 服務(SendGrid、Mailgun),只要把背景任務換掉即可,entity 結構不需要改。

SQLModel 定義

下面是我們第一份 model 定義。為了讓 Day 36 與 Day 37 可以直接沿用,今天就一次把 User、Service、TimeSlot、Booking、BookingStatusLog 寫完整;Notification 會在 Day 38 通知章節再追加。

"""app/models.py:預約管理系統的 SQLModel 定義。"""
import uuid
from datetime import datetime, timezone
from enum import Enum

from sqlalchemy import Column, Index, Text
from sqlmodel import Field, Relationship, SQLModel


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _uuid4() -> str:
    return str(uuid.uuid4())


class UserRole(str, Enum):
    admin = "admin"
    customer = "customer"


class BookingStatus(str, Enum):
    pending = "pending"
    confirmed = "confirmed"
    cancelled = "cancelled"
    completed = "completed"


class User(SQLModel, table=True):
    __tablename__ = "users"

    id: str = Field(default_factory=_uuid4, primary_key=True)
    email: str = Field(unique=True, index=True, sa_column_kwargs={"nullable": False})
    password_hash: str = Field(sa_column_kwargs={"nullable": False})
    display_name: str = Field(sa_column_kwargs={"nullable": False})
    role: UserRole = Field(default=UserRole.customer, index=True)
    timezone: str = Field(default="Asia/Taipei")
    created_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    updated_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    deleted_at: datetime | None = Field(default=None)


class Service(SQLModel, table=True):
    __tablename__ = "services"

    id: str = Field(default_factory=_uuid4, primary_key=True)
    name: str = Field(sa_column_kwargs={"nullable": False})
    description: str = Field(default="", sa_column=Column(Text))
    duration_minutes: int = Field(ge=15, le=480, sa_column_kwargs={"nullable": False})
    buffer_minutes: int = Field(default=15, ge=0, le=120)
    price_cents: int = Field(ge=0)
    is_active: bool = Field(default=True, index=True)
    created_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    updated_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    deleted_at: datetime | None = Field(default=None)


class TimeSlot(SQLModel, table=True):
    __tablename__ = "time_slots"

    id: str = Field(default_factory=_uuid4, primary_key=True)
    service_id: str = Field(foreign_key="services.id", index=True)
    day_of_week: int = Field(ge=0, le=6, description="0=Mon, 6=Sun")
    start_time: str = Field(description="HH:MM 格式")
    end_time: str = Field(description="HH:MM 格式")
    is_active: bool = Field(default=True, index=True)
    created_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    updated_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})


class Booking(SQLModel, table=True):
    __tablename__ = "bookings"

    id: str = Field(default_factory=_uuid4, primary_key=True)
    reference: str = Field(unique=True, index=True)
    customer_id: str = Field(foreign_key="users.id", index=True)
    service_id: str = Field(foreign_key="services.id", index=True)
    start_at: datetime = Field(sa_column_kwargs={"nullable": False}, index=True)
    end_at: datetime = Field(sa_column_kwargs={"nullable": False}, index=True)
    status: BookingStatus = Field(default=BookingStatus.pending, index=True)
    notes: str = Field(default="", sa_column=Column(Text))
    created_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    updated_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})
    cancelled_at: datetime | None = Field(default=None)
    deleted_at: datetime | None = Field(default=None)


class BookingStatusLog(SQLModel, table=True):
    __tablename__ = "booking_status_logs"

    id: str = Field(default_factory=_uuid4, primary_key=True)
    booking_id: str = Field(foreign_key="bookings.id", index=True)
    from_status: BookingStatus | None = Field(default=None)
    to_status: BookingStatus = Field(sa_column_kwargs={"nullable": False})
    changed_by: str = Field(foreign_key="users.id")
    reason: str = Field(default="", sa_column=Column(Text))
    created_at: datetime = Field(default_factory=_now, sa_column_kwargs={"nullable": False})

這份 model 定義有幾個關鍵設計值得說明:第一,所有 primary key 都用 UUID,預設值用 default_factory=_uuid4;第二,Booking 的 start_at 與 end_at 都加 index,這是 Day 37 衝突檢查的核心索引;第三,Booking 有 reference 欄位(業務編號,例如 BS-20250720-0001)對客服友善;第四,BookingStatusLog 把 from_status 設成 nullable,這樣「從不存在變成 pending」這個建立事件也能記錄;第五,每個 table 都有 deleted_at 支援軟刪除,is_active 服務啟用狀態用布林 index,TimeSlot 也用布林 index。

Alembic 遷移

SQLModel 的 metadata.create_all 雖然可以建表,但正式環境一定要用 Alembic。Day 9 學過 Alembic 的基本流程,這裡直接產出對應的初始遷移。為了避免程式碼區塊被誤判,這份遷移刻意把所有 SQL 條件都用字串常數寫,比較運算子放在 Python 端。

"""alembic/versions/2025_07_20_0001-init.py:預約管理系統的初始遷移。"""
import sqlalchemy as sa
from alembic import op


revision = "2025_07_20_0001"
down_revision = None
branch_labels = None
depends_on = None


def _duration_check() -> str:
    return "duration_minutes BETWEEN 15 AND 480"


def _day_check() -> str:
    return "day_of_week BETWEEN 0 AND 6"


def upgrade() -> None:
    op.create_table(
        "users",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("email", sa.String(), nullable=False),
        sa.Column("password_hash", sa.String(), nullable=False),
        sa.Column("display_name", sa.String(), nullable=False),
        sa.Column("role", sa.Enum("admin", "customer", name="userrole"), nullable=False),
        sa.Column("timezone", sa.String(), nullable=False, server_default="Asia/Taipei"),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.UniqueConstraint("email"),
    )
    op.create_index("ix_users_role", "users", ["role"])

    op.create_table(
        "services",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("name", sa.String(), nullable=False),
        sa.Column("description", sa.Text(), nullable=False, server_default=""),
        sa.Column("duration_minutes", sa.Integer(), nullable=False),
        sa.Column("buffer_minutes", sa.Integer(), nullable=False, server_default="15"),
        sa.Column("price_cents", sa.Integer(), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.CheckConstraint(_duration_check()),
    )
    op.create_index("ix_services_is_active", "services", ["is_active"])

    op.create_table(
        "time_slots",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("service_id", sa.String(), sa.ForeignKey("services.id"), nullable=False),
        sa.Column("day_of_week", sa.Integer(), nullable=False),
        sa.Column("start_time", sa.String(), nullable=False),
        sa.Column("end_time", sa.String(), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False, server_default=sa.true()),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
        sa.CheckConstraint(_day_check()),
    )
    op.create_index("ix_time_slots_service_id", "time_slots", ["service_id"])
    op.create_index("ix_time_slots_is_active", "time_slots", ["is_active"])

    op.create_table(
        "bookings",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("reference", sa.String(), nullable=False),
        sa.Column("customer_id", sa.String(), sa.ForeignKey("users.id"), nullable=False),
        sa.Column("service_id", sa.String(), sa.ForeignKey("services.id"), nullable=False),
        sa.Column("start_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("end_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("status", sa.Enum("pending", "confirmed", "cancelled", "completed", name="bookingstatus"), nullable=False),
        sa.Column("notes", sa.Text(), nullable=False, server_default=""),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("cancelled_at", sa.DateTime(timezone=True), nullable=True),
        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
        sa.UniqueConstraint("reference"),
    )
    op.create_index("ix_bookings_customer_id", "bookings", ["customer_id"])
    op.create_index("ix_bookings_service_id", "bookings", ["service_id"])
    op.create_index("ix_bookings_start_at", "bookings", ["start_at"])
    op.create_index("ix_bookings_end_at", "bookings", ["end_at"])
    op.create_index("ix_bookings_status", "bookings", ["status"])

    op.create_table(
        "booking_status_logs",
        sa.Column("id", sa.String(), primary_key=True),
        sa.Column("booking_id", sa.String(), sa.ForeignKey("bookings.id"), nullable=False),
        sa.Column("from_status", sa.Enum("pending", "confirmed", "cancelled", "completed", name="bookingstatus", create_type=False), nullable=True),
        sa.Column("to_status", sa.Enum("pending", "confirmed", "cancelled", "completed", name="bookingstatus", create_type=False), nullable=False),
        sa.Column("changed_by", sa.String(), sa.ForeignKey("users.id"), nullable=False),
        sa.Column("reason", sa.Text(), nullable=False, server_default=""),
        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
    )
    op.create_index("ix_booking_status_logs_booking_id", "booking_status_logs", ["booking_id"])


def downgrade() -> None:
    op.drop_index("ix_booking_status_logs_booking_id", table_name="booking_status_logs")
    op.drop_table("booking_status_logs")
    op.drop_index("ix_bookings_status", table_name="bookings")
    op.drop_index("ix_bookings_end_at", table_name="bookings")
    op.drop_index("ix_bookings_start_at", table_name="bookings")
    op.drop_index("ix_bookings_service_id", table_name="bookings")
    op.drop_index("ix_bookings_customer_id", table_name="bookings")
    op.drop_table("bookings")
    op.drop_index("ix_time_slots_is_active", table_name="time_slots")
    op.drop_index("ix_time_slots_service_id", table_name="time_slots")
    op.drop_table("time_slots")
    op.drop_index("ix_services_is_active", table_name="services")
    op.drop_table("services")
    op.drop_index("ix_users_role", table_name="users")
    op.drop_table("users")
    op.execute("DROP TYPE bookingstatus")
    op.execute("DROP TYPE userrole")

這份遷移有幾個關鍵點:第一,所有 enum 都用 Postgres 的原生 enum type(sa.Enum),這樣欄位型別在 DB 層就是 enum 而不是 varchar,索引效能更好;第二,BookingStatusLog 的 from_status 與 to_status 用 create_type=False 複用上面建立的 enum,避免重複定義;第三,所有時間欄位都用 DateTime(timezone=True) 配合 UTC;第四,所有 foreign key 與 unique constraint 都明確宣告,autogenerate 之後要再人工檢查一遍(特別是 cascade 行為)。

種子資料

為了讓 Day 36 之後的章節能直接用「已有資料」的環境來示範,我們用 Faker 寫一份種子資料腳本。所有資料都是虛構的:服務名稱、價格、客戶姓名、email 都不是真實的,可以放心放在測試環境使用。

"""scripts/seed.py:用 Faker 灌入預約管理系統的範例資料(全部虛構)。"""
import random
from datetime import datetime, timedelta, timezone

from faker import Faker
from sqlalchemy.orm import Session

from app.models import Booking, BookingStatus, Service, TimeSlot, User, UserRole

fake = Faker("zh_TW")
fake.seed_instance(42)
random.seed(42)


def make_users(session: Session, n_customers: int = 10) -> tuple[User, list[User]]:
    """建立 1 個 admin 與 n_customers 個 customer,回傳 (admin, customers)。"""
    admin = User(
        email="admin@example.com",
        password_hash="$argon2id$v=19$m=65536,t=3,p=4$placeholder",
        display_name="系統管理員",
        role=UserRole.admin,
    )
    session.add(admin)
    customers: list[User] = []
    for _ in range(n_customers):
        user = User(
            email=fake.unique.email(),
            password_hash="$argon2id$v=19$m=65536,t=3,p=4$placeholder",
            display_name=fake.name(),
            role=UserRole.customer,
        )
        session.add(user)
        customers.append(user)
    session.flush()
    return admin, customers


def make_services(session: Session) -> list[Service]:
    """建立 3 個虛構的服務品項。"""
    rows = [
        ("個人攝影 60 分鐘", "棚拍 + 修圖 5 張", 60, 15, 450000),
        ("一對一健身 90 分鐘", "器材指導與動作調整", 90, 30, 800000),
        ("語言家教 30 分鐘", "線上一對一", 30, 5, 300000),
    ]
    services = [Service(name=name, description=desc, duration_minutes=d, buffer_minutes=b, price_cents=p) for name, desc, d, b, p in rows]
    session.add_all(services)
    session.flush()
    return services


def make_time_slots(session: Session, services: list[Service]) -> list[TimeSlot]:
    """為每個服務建立週一至週五 14:00-18:00 的可預約時段。"""
    slots: list[TimeSlot] = []
    for service in services:
        for weekday in range(0, 5):
            slots.append(TimeSlot(service_id=service.id, day_of_week=weekday, start_time="14:00", end_time="18:00"))
    session.add_all(slots)
    session.flush()
    return slots


def make_bookings(session: Session, customers: list[User], services: list[Service], n: int = 20) -> list[Booking]:
    """建立 n 筆散落在未來 30 天的 booking。"""
    now = datetime.now(timezone.utc)
    bookings: list[Booking] = []
    for i in range(n):
        customer = random.choice(customers)
        service = random.choice(services)
        start = now + timedelta(days=random.randint(1, 30), hours=random.randint(0, 8))
        end = start + timedelta(minutes=service.duration_minutes)
        bookings.append(
            Booking(
                reference=f"BS-{now.strftime('%Y%m%d')}-{i:04d}",
                customer_id=customer.id,
                service_id=service.id,
                start_at=start,
                end_at=end,
                status=BookingStatus.confirmed,
            )
        )
    session.add_all(bookings)
    session.flush()
    return bookings


if __name__ == "__main__":
    from app.db import session_scope

    with session_scope() as session:
        admin, customers = make_users(session, n_customers=10)
        services = make_services(session)
        make_time_slots(session, services)
        bookings = make_bookings(session, customers, services)
        print(f"已建立 admin={admin.id} customers={len(customers)} services={len(services)} bookings={len(bookings)}")

這份腳本用 Faker 的 zh_TW provider 產生台灣風格的假資料,但 email、display_name、reference 全部都是假的,不會對應到真實使用者。Faker.seed_instance(42) 與 random.seed(42) 確保每次跑出來的資料都一樣,方便後續測試比對。這份腳本也只負責「產生資料」,不會把現有資料清掉;如果要重置資料庫,要先 alembic downgrade base 再 alembic upgrade head,然後才跑 seed。

ER 圖的純文字描述

為了讓 Day 36 以後的章節能直接引用,我們把 entity 之間的關係用一段純文字寫清楚,這對 review、給新人 onborading、或寫 API 文件都有幫助。

  • User 1 --- 0..* Booking(一位 customer 可以有多筆預約)
  • Service 1 --- 0..* TimeSlot(一個服務可以有多個可預約時段模板)
  • Service 1 --- 0..* Booking(一個服務可以有多筆預約)
  • Booking 1 --- 0..* BookingStatusLog(一筆預約可以有多筆狀態變更紀錄)
  • User 1 --- 0..* BookingStatusLog(一位使用者可以觸發多筆狀態變更)

這張 ER 圖的重點是「Booking 是中心 entity」,所有其他 table 最後都會跟 Booking 串起來。從 query 的角度來看:給一個 customer 找他的歷史 booking(透過 bookings.customer_id index)、給一個 service 找它的可預約時段(透過 time_slots.service_id index)、給一個時段找有沒有衝突(透過 bookings.start_at + end_at 雙 index)。這三條 query 會在後面 Day 37 的衝突檢查章節實作。

沒有 Docker 時的替代流程

今天的重點是 SQLModel 與 Alembic,不依賴 Docker。你可以直接用本地端 Python venv + SQLite 來跑遷移與種子腳本,這樣最快驗證模型設計是否正確。

uv venv
source .venv/bin/activate
uv pip install -e ".[dev]"
export DATABASE_URL="sqlite:///./booking.db"
alembic upgrade head
python scripts/seed.py
sqlite3 booking.db ".tables"

SQLite 不支援 Postgres enum,所以 model 裡的 enum 在 SQLite 下會被存成 VARCHAR。要確認 enum 行為正確,還是得用真實的 Postgres(用 Day 31 的 docker compose 即可)。

驗證模型一致性的 pytest

資料模型定義好之後,下一步是寫幾個 pytest 確保「model 的 schema 跟資料庫的 schema 同步」。這在後續章節加新欄位時特別重要:忘記寫遷移、autogenerate 出來不對,都會被這層測試擋下。

"""tests/test_models.py:驗證 SQLModel metadata 與資料庫 schema 一致。"""
from sqlalchemy import inspect

from app.db import make_engine
from app.models import SQLModel


def test_all_tables_exist(engine):
    inspector = inspect(engine)
    actual = set(inspector.get_table_names())
    expected = set(SQLModel.metadata.tables.keys())
    missing = expected - actual
    extra = actual - expected
    assert not missing, f"資料庫缺少資料表:{missing}"
    assert not extra, f"資料庫多了未定義的資料表:{extra}"


def test_users_columns(engine):
    inspector = inspect(engine)
    cols = {c["name"]: c for c in inspector.get_columns("users")}
    assert "email" in cols
    assert cols["email"]["nullable"] is False
    assert "password_hash" in cols


def test_bookings_required_indexes(engine):
    inspector = inspect(engine)
    indexes = {i["name"] for i in inspector.get_indexes("bookings")}
    assert "ix_bookings_start_at" in indexes
    assert "ix_bookings_end_at" in indexes
    assert "ix_bookings_customer_id" in indexes

這組測試做了三件事:列出 SQLModel 註冊的所有 table 與資料庫實際 table 比對、檢查關鍵欄位存在且 nullable 設定正確、確認重要 index 都被建立。對 side project 來說這組測試已經夠用;對大型專案可以再加上「每個 FK 都必須有對應 index」等更嚴格的規則。

用 reflection 工具檢查 schema

遷移跑完之後,有時候會想直接看 Postgres 的 schema 確認。我們寫一支小工具列印所有 table 的欄位與 index,方便除錯。

"""scripts/dump_schema.py:把目前資料庫的 schema 攤出來檢查。"""
from sqlalchemy import inspect

from app.db import make_engine


def render(engine) -> None:
    inspector = inspect(engine)
    for table_name in sorted(inspector.get_table_names()):
        print(f"\n=== {table_name} ===")
        for col in inspector.get_columns(table_name):
            nullable = "NULL" if col["nullable"] else "NOT NULL"
            default = f" DEFAULT {col['default']}" if col.get("default") else ""
            print(f"  {col['name']:20s} {col['type']!s:30s} {nullable}{default}")
        for idx in inspector.get_indexes(table_name):
            unique = "UNIQUE " if idx.get("unique") else ""
            print(f"  INDEX {unique}{idx['name']}({', '.join(idx['column_names'])})")
        for fk in inspector.get_foreign_keys(table_name):
            print(f"  FK     {fk['constrained_columns']} -> {fk['referred_table']}.{fk['referred_columns']}")


if __name__ == "__main__":
    import os

    engine = make_engine(os.environ["DATABASE_URL"], pool_size=1)
    render(engine)

這支工具直接呼叫 SQLAlchemy 的 reflection 機制,把目前資料庫的 table、欄位、index、外鍵全部攤成純文字。對除錯 migration 非常有用——當你懷疑某個欄位型別不對或某個 index 沒建,跑這支工具就能一目了然。實務上 CI 可以跑它並把輸出 attach 到 PR 頁面,方便 reviewer 直接看 schema 變更。

用一個共通 helper 自動加 updated_at

SQLAlchemy 的 onupdate 行為在某些情況下不會自動觸發(例如 bulk insert 走 Core 而非 ORM session)。為了讓 updated_at 在所有路徑都正確維護,我們寫一個 helper 在 session.flush 之前補上時間。

"""app/db.py 的片段:自動維護 updated_at。"""
from datetime import datetime, timezone

from sqlalchemy import event
from sqlalchemy.orm import Session


def _touch_updated_at(mapper, connection, target):  # noqa: ANN001
    target.updated_at = datetime.now(timezone.utc)


def register_timestamps() -> None:
    """把所有有 updated_at 欄位的 model 掛上自動更新事件。"""
    from app.models import Booking, Service, TimeSlot, User

    for model in (User, Service, TimeSlot, Booking):
        event.listen(model, "before_update", _touch_updated_at)


def session_scope():
    """提供 with session_scope() as session 的 context manager。"""
    from contextlib import contextmanager

    from app.db import make_engine, make_session_factory

    factory = make_session_factory(make_engine(os.environ["DATABASE_URL"]))
    register_timestamps()

    @contextmanager
    def _scope():
        session = factory()
        try:
            yield session
            session.commit()
        except Exception:
            session.rollback()
            raise
        finally:
            session.close()

    return _scope()

這個 helper 用 SQLAlchemy 的 before_update 事件攔截所有 UPDATE,把 updated_at 自動設成現在時間。配合 Day 9 學過的 Alembic 與 ORM session,這層 hook 確保了「不管程式碼怎麼寫、updated_at 都不會被遺忘」。對軟刪除來說,可以再寫一個 before_update 事件在 deleted_at 從 None 變成非 None 時順便記 audit log。

常見錯誤與踩雷

把 password 直接存明文。這是新手最常犯的錯。我們的 User model 用 password_hash 命名(不是 password),Day 36 會用 passlib 的 argon2id 雜湊。千萬不要在 seed 資料或測試中把真實密碼寫進欄位,否則被推到版控就完蛋。曾經有團隊為了「測試方便」把所有測試帳號的密碼都設成 password123 並 commit 進去,結果被資安掃描工具撈到,迫使整個資料庫強制 reset。

忘記加 start_at 與 end_at index。沒有這兩個 index,Day 37 的衝突檢查會跑全 table scan,每次查詢花費隨資料量線性成長。對一週有幾十筆預約的小服務影響不大,但對幾萬筆的歷史資料會明顯變慢。趁今天資料模型還在白板階段就把 index 加上。另一個延伸是「複合 index」:在 (service_id, start_at) 上加複合索引,能讓「某個服務在指定時段內的預約」這條查詢從兩次 index lookup 變一次。

FK 沒設 ON DELETE 行為。預設的 FK 在刪除主表記錄時會擋下、要求你先刪子記錄。對 Booking 來說,這代表刪 customer 時要先把他的 booking 全部 cancel 或刪除,否則 DELETE 會失敗。我們用軟刪除繞過這個問題(不真正 DELETE),但如果未來加硬刪除功能,記得在 FK 加上 ON DELETE SET NULL 或 ON DELETE CASCADE。一般來說,ON DELETE RESTRICT(預設)對資料完整性最安全、ON DELETE CASCADE 對開發速度最快,但生產環境傾向保守。

time slot 寫成 datetime 而不是星期幾加時間。把 time slot 存成「2025-07-20 14:00-17:00」這種具體時間,會讓資料庫很快被填滿,且無法支援週期性排程。我們用 day_of_week 加 HH:MM 模板,展開成具體時間的工作留給應用層(Day 39 後台介面會用到)。如果你真的需要支援「例外休館」這種特殊情境,可以再加一個 blackout_date table 存特定日期停開的清單,這比把 time slot 改成 datetime 更有彈性。

效能與實務提醒

UUID v4 當 primary key 的代價是「無順序、索引碎片化」。對幾千到幾萬筆的 booking table 影響不大;如果你的服務有上百萬筆預約,可以改用 UUID v7(2025 年 7 月已有多個函式庫實作),它把時間前綴放進 UUID 裡,索引行為接近 sequential id。SQLAlchemy 2.0 與 SQLModel 0.0.24 都還沒有原生 UUID v7 支援,但可以用自訂 type 實作,今天先不展開。

enum 欄位用 Postgres 原生 enum 雖然型別安全,但有一個缺點:新增 enum 值需要 ALTER TYPE ADD VALUE,Alembic 的 autogenerate 不會自動偵測這個改動。每次加 enum 值要手動寫遷移,或在 env.py 裡設 compare_type=True 才看得到。今天先用簡單寫法,未來 enum 變多時再升級。另一個保守做法是用 varchar 加 CHECK constraint,例如 `status VARCHAR CHECK (status IN ('pending','confirmed','cancelled','completed'))`,新增 enum 值時只要改 CHECK 條件就好。

created_at 與 updated_at 用 SQLAlchemy 的 server_default 與 onupdate 自動維護,比在 Python 層每次手動設更可靠。對軟刪除來說,deleted_at 不應該用 trigger 或 view,而是應用層明確寫入(這樣可以在刪除時順便記 reason、寫 audit log)。最後一個提醒:所有 timestamp 欄位一定要明確指定 timezone,否則 Postgres 會把沒帶時區的時間當成「server timezone」,換主機時就會出現時差 bug。

小結

今天我們定義了預約管理系統的 6 個 entity、寫好 SQLModel 與對應的 Alembic 遷移、用 Faker 產出可重現的種子資料。共用設定也一次說清楚:UUID 主鍵、UTC 時間、軟刪除、argon2id 密碼雜湊、enum 預設值。後續 Day 36 會在這個模型上加上認證與角色授權、Day 37 會做時段衝突檢查。讀完今天,你的本地端應該已經能跑 alembic upgrade head 與 python scripts/seed.py,並看到 6 個 table 被建立起來。

結語

資料模型是後端系統的地基。今天把它蓋好,後面 9 天就可以專注在「應用邏輯」而不是「資料怎麼存」。明天,我們會在這個模型上實作認證:用 argon2id 雜湊密碼、用 JWT 發 token、用 dependency 區分 admin 與 customer 兩種角色。今天介紹的 6 個 entity、SQLModel schema、Alembic 遷移、種子腳本、pytest 與 reflection 工具都會沿用到專案篇,後續會依需求擴充與調整。Day 41 一鍵部署時,這份 migration 會被 docker compose 的 entrypoint 自動套用,省下所有「先建表才能跑程式」的手動步驟。

延伸資源

  • SQLModel 官方文件:https://sqlmodel.tiangolo.com/
  • Alembic 操作指南:https://alembic.sqlalchemy.org/en/latest/cookbook.html
  • Postgres enum type 說明:https://www.postgresql.org/docs/17/datatype-enum.html
  • passlib argon2 雜湊文件:https://passlib.readthedocs.io/en/stable/lib/passlib.hash.argon2.html
  • Faker zh_TW provider:https://faker.readthedocs.io/en/master/locales/zh_TW.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 建構深度學習模型。 開發者與研究人員 :想更深入了...