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
留言
張貼留言