Web Day 9 遷移管理:Alembic
執行需求:CPU 可跑。昨天我們在 SQLModel 上做了一對多與多對多的關聯設計,現在資料表變多了;只要改一個欄位或加一張表,整個資料庫結構就跟著改。前幾天我們都靠 SQLModel.metadata.create_all 重建結構,這在開發階段沒問題,但只要部署到正式環境或團隊合作時,這個方法就會捅出大簍子。今天我們要引進 Alembic 這套業界標準的資料庫遷移(migration)工具,把「結構變更」從「人工作業」升級成「可版本化、可審查、可回溯」的流程。
引言
寫後端的工程師都遇過這種情境:「欸,我這邊跑起來沒事啊?」,問題出在你改了資料模型,但同事的資料庫還是舊結構;或正式環境還沒拿到新欄位,舊程式讀不到值就壞掉。資料庫結構不像程式碼可以用 Git 一鍵同步,它是「狀態」,必須靠「遷移檔案」一段一段往前推,而且每個環境(開發、測試、正式)都要套用同一套變更。Alembic 就是處理這件事最普遍的工具,SQLAlchemy 官方推薦它,FastAPI + SQLModel 生態也把它當作預設的遷移方案。
今天的目標有四個:第一,理解「為什麼需要遷移管理」以及它解決了哪些問題;第二,安裝並初始化 Alembic;第三,學會用 Alembic 偵測 SQLModel 的模型變更並產生遷移檔;第四,學會用 alembic upgrade、alembic downgrade 在 SQLite 上完整跑一次升級與回退。讀完之後,你會知道怎麼在「改了 SQLModel 模型」之後讓資料庫跟上,並且能在出問題時安全地回到上一個版本。
這篇文章預設你已經會用 SQLModel 定義模型(Web Day 6)、會用 Session 與 Engine 操作 SQLite(Web Day 6、Web Day 7)、並且了解關聯設計的基本觀念(Web Day 8)。如果對這些主題還不熟,建議先回頭看那三篇再回來,否則會跟不上範例的節奏。
為什麼需要遷移管理
資料庫的「結構」跟「資料」是兩件事。資料可以被丟掉再重建,但結構一旦設計錯,往往要花很多力氣回頭修。在小型專案裡,我們可能會用 SQLModel.metadata.create_all(engine) 在應用程式啟動時自動建表:第一次跑會建立所有資料表,第二次跑則不會動既有的表。這種做法在開發階段沒問題,但有以下限制:第一,它只能「建立」,不能「修改欄位」(例如把 VARCHAR(50) 改成 VARCHAR(200));第二,它不能「刪除欄位」;第三,部署到正式環境時,應用程式啟動時自動改結構非常危險,因為你無法控制時機、也無法審查變更內容。
遷移管理把「結構變更」變成「程式碼」。每次要改資料表,就寫一支遷移腳本(migration script),內容是「從版本 A 升級到版本 B 的具體步驟」,例如「新增 price 欄位,型別是 NUMERIC(10,2),預設值 0」。這支腳本會被簽入版控,所有人、所有環境都用同一支腳本升級。出問題時,還能寫一支「降級」腳本,把結構倒回去。這個流程在業界叫做「schema migration」,Alembic 是 Python 生態最廣泛使用的工具。
Alembic 是 SQLAlchemy 的姊妹專案,作者是同一位(Mike Bayer)。它的核心概念是「每支遷移檔都有一個版本編號」,資料庫裡有一張 alembic_version 表,記錄目前在哪個版本。執行 alembic upgrade head 時,Alembic 會讀這張表,套用所有還沒跑過的遷移檔,直到抵達最新版(head)。這聽起來簡單,實際運作起來則牽涉到模型比對、SQL 產生、交易控制、自動 vs 手動遷移等細節。
SQLModel 雖然是基於 SQLAlchemy,但它的 metadata 是「繼承自 SQLAlchemy 的 MetaData」,所以 Alembic 能直接讀 SQLModel 定義的模型來產生遷移。這是為什麼我們選擇 SQLModel 而非單獨 SQLAlchemy 的好處之一:遷移工具鏈完全相容。今天我們就用 SQLModel 的既有模型做示範。
安裝 Alembic 並初始化
先把專案環境整理好。我們沿用 Web Day 2 的專案結構,pyproject.toml 已經管 FastAPI、SQLModel、uvicorn。現在加入 Alembic 1.16(2025 年 7 月的主流版本)。
在命令列執行(這是 bash 指令,不是 Python):
# 加入 Alembic 到依賴
# 使用 uv 範例(Web Day 2 介紹過)
uv add "alembic==1.16"
# 或使用 pip
pip install "alembic==1.16"
接下來建立一個簡化的 SQLModel 模型當作今天的範例。檔案 app/models.py 沿用 Web Day 6–8 的寫法,定義 Hero 與 Team 兩張表(一對多關聯):
# app/models.py
# 今天的範例沿用 Web Day 8 的 Hero 與 Team 關聯
from datetime import datetime
from typing import Optional
from sqlmodel import Field, Relationship, SQLModel
class Team(SQLModel, table=True):
# 團隊表:一個團隊有多位英雄
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True, unique=True)
headquarters: str
heroes: list["Hero"] = Relationship(back_populates="team")
class Hero(SQLModel, table=True):
# 英雄表:每位英雄屬於一個團隊
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None)
# Web Day 8 示範用的時間戳記欄位
created_at: datetime = Field(default_factory=datetime.utcnow)
team_id: Optional[int] = Field(default=None, foreign_key="team.id")
team: Optional[Team] = Relationship(back_populates="heroes")
這個檔案定義了兩張表:Team 有 id、name、headquarters,Hero 有 id、name、secret_name、age、created_at,並透過 team_id 與 Team 關聯。Alembic 之後會讀 SQLModel 內部的 metadata,把這些欄位跟型別翻譯成對應的 SQL。
現在初始化 Alembic。在專案根目錄執行:
# 在專案根目錄執行(app/ 與 pyproject.toml 同層)
alembic init alembic
# 輸出範例:
# Creating directory /path/to/project/alembic ... done
# Creating directory /path/to/project/alembic/versions ... done
# Generating /path/to/project/alembic/env.py ... done
# Generating /path/to/project/alembic/script.py.mako ... done
# Generating /path/to/project/alembic/README ... done
# Generating /path/to/project/alembic.ini ... done
# Please edit configuration/connection/logging settings here before running.
alembic init 會建立一個 alembic/ 目錄與一個 alembic.ini 設定檔。alembic/versions/ 是放遷移檔的地方;alembic/env.py 是執行遷移時實際跑的程式,Alembic 在這個檔案裡讀取資料庫連線資訊與 metadata。我們等一下要改 env.py,讓它讀 SQLModel 的 metadata。
連線設定:把 Alembic 接到 SQLModel
Alembic 預設會從 alembic.ini 讀資料庫網址(sqlalchemy.url)。但直接把網址寫死在 ini 檔有兩個缺點:第一,不同環境(開發、測試、正式)需要不同網址;第二,網址裡通常含有密碼,簽進版控會洩漏。慣例做法是讓 env.py 從環境變數或應用程式的設定讀網址。
打開 alembic/env.py,把原本的 target_metadata 換成 SQLModel 的 metadata,並設定 sqlalchemy.url:
# alembic/env.py(重點段落)
# 從你的應用程式 import metadata,這樣 Alembic 才能比對模型
import os
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
from sqlmodel import SQLModel
# 假設 app.models 模組裡定義了所有 SQLModel 模型
# import 之後 SQLModel.metadata 會自動收集它們
from app.models import Hero, Team # noqa: F401
config = context.config
# 從環境變數讀取 DATABASE_URL;若沒設定則用 SQLite 本機檔案
DATABASE_URL = os.environ.get(
"DATABASE_URL", "sqlite:///./app.db"
)
config.set_main_option("sqlalchemy.url", DATABASE_URL)
# 讓 Alembic 用 SQLModel 的 metadata 作為比對基準
target_metadata = SQLModel.metadata
if config.config_file_name is not None:
fileConfig(config.config_file_name)
這段做了三件事:把 target_metadata 指向 SQLModel.metadata,這樣 Alembic 才能比對「資料庫目前的結構」與「模型定義的結構」;從環境變數讀資料庫網址,未來切換正式環境時只要改環境變數;import 模型模組(from app.models import Hero, Team)是必要的,否則 metadata 會是空的,自動偵測也抓不到東西。
如果你的專案結構稍有不同(例如模型放在 app/db/models.py),把 import 路徑換成對應的位置就好。重點是:import 之後,SQLModel.metadata 才會收集到模型,這是 SQLAlchemy 的標準機制。
第一支遷移:自動偵測 vs 手動撰寫
Alembic 提供兩種產生遷移檔的方式。第一種是「自動偵測」:執行 alembic revision --autogenerate -m "訊息",Alembic 會比對模型與資料庫,自動產生遷移腳本。第二種是「手動撰寫」:執行 alembic revision -m "訊息" 產生空腳本,自己寫升級與降級邏輯。實務上多半兩者並行:日常欄位異動用 autogenerate,特殊資料遷移(例如把舊欄位的值搬到新欄位)用手寫。
我們先用自動偵測做第一支遷移。第一次跑之前,要先把 SQLModel 模型的「目標狀態」建立起來,但資料庫目前是空的(或舊結構)。執行:
# 產生第一支遷移檔(autogenerate)
alembic revision --autogenerate -m "init hero and team"
# 輸出範例:
# Generating /path/to/alembic/versions/abc123_init_hero_and_team.py ... done
打開剛產生的遷移檔,你會看到類似這樣的內容:
# alembic/versions/abc123_init_hero_and_team.py
"""init hero and team
Revision ID: abc123def456
Revises:
Create Date: 2025-07-21 10:00:00.000000
"""
from typing import Sequence, Union
import sqlalchemy as sa
import sqlmodel
from alembic import op
revision: str = "abc123def456"
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# 自動產生的升級動作
op.create_table(
"team",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.Column("headquarters", sa.String(), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("name"),
)
op.create_index("ix_team_name", "team", ["name"], unique=True)
op.create_table(
"hero",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(), nullable=False),
sa.Column("secret_name", sa.String(), nullable=False),
sa.Column("age", sa.Integer(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=False),
sa.Column("team_id", sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(["team_id"], ["team.id"]),
sa.PrimaryKeyConstraint("id"),
)
op.create_index("ix_hero_name", "hero", ["name"], unique=False)
def downgrade() -> None:
# 自動產生的降級動作
op.drop_index("ix_hero_name", table_name="hero")
op.drop_table("hero")
op.drop_index("ix_team_name", table_name="team")
op.drop_table("team")
upgrade() 是「從舊版升級到新版」的動作;downgrade() 是反過來。每一支遷移檔都有一個 revision 識別碼與 down_revision(前一版的識別碼),形成一條版本鏈。Alembic 靠這條鏈決定要從哪裡開始套用。自動產生的腳本對「建表」這種場景非常準確,但對「改欄位型別」「刪除欄位」要特別小心:autogenerate 不會主動幫你搬資料,必須手動補上 op.execute(...)。
執行遷移與驗證
產生遷移檔之後,把它套用到資料庫:
# 把所有未套用的遷移升級到最新版
alembic upgrade head
# 輸出範例:
# INFO [alembic.runtime.migration] Context impl SQLiteImpl.
# INFO [alembic.runtime.migration] Will assume non-transactional DDL.
# INFO [alembic.runtime.migration] Running upgrade -> abc123, init hero and team
執行完成後,SQLite 檔案 app.db 會出現三張表:team、hero、alembic_version。第三張是 Alembic 內部用的版本表,用來記錄目前在哪個版本。我們可以用 SQLite 命令列工具或 SQLModel 開 session 驗證:
# verify_alembic.py
# 驗證遷移後的資料表結構
from sqlmodel import Session, SQLModel, create_engine, select
from app.models import Hero, Team
engine = create_engine("sqlite:///./app.db")
with Session(engine) as session:
# 驗證 team 表是空的、可以新增
session.add(Team(name="Avengers", headquarters="New York"))
session.add(Team(name="Justice League", headquarters="Metropolis"))
session.commit()
teams = session.exec(select(Team)).all()
print(f"團隊數量:{len(teams)}")
# 輸出:團隊數量:2
# 驗證 hero 表與 team 關聯
avengers = session.exec(
select(Team).where(Team.name == "Avengers")
).one()
session.add(
Hero(
name="Spider-Man",
secret_name="Peter Parker",
age=23,
team_id=avengers.id,
)
)
session.commit()
heroes = session.exec(select(Hero)).all()
print(f"英雄數量:{len(heroes)}")
# 輸出:英雄數量:1
這段先用 select(Team) 取出所有團隊,確認遷移後的表能正常查詢與新增;接著新增一位英雄並帶上 team_id,確認外鍵運作正常。如果兩段都沒錯誤,代表遷移成功。
改模型再遷移:第二支
接下來示範「改了模型,怎麼再產生遷移」。我們在 Hero 上加一個 power_level 欄位(戰力值,用整數表示):
# app/models.py(修改後的 Hero)
class Hero(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = Field(index=True)
secret_name: str
age: Optional[int] = Field(default=None)
created_at: datetime = Field(default_factory=datetime.utcnow)
# 新增欄位:戰力值
power_level: int = Field(default=100)
team_id: Optional[int] = Field(default=None, foreign_key="team.id")
team: Optional[Team] = Relationship(back_populates="heroes")
改完模型後,再跑一次自動偵測:
alembic revision --autogenerate -m "add hero power_level"
# 輸出範例:
# Generating /path/to/alembic/versions/def456_add_hero_power_level.py ... done
打開新產生的檔案,Alembic 應該會聰明地只補上「新增欄位」的動作,不會動到既有的欄位:
# alembic/versions/def456_add_hero_power_level.py
def upgrade() -> None:
# Alembic 偵測到新增欄位,自動補上 addcolumn
op.add_column(
"hero",
sa.Column(
"power_level",
sa.Integer(),
nullable=False,
server_default="100",
),
)
def downgrade() -> None:
op.drop_column("hero", "power_level")
注意 server_default="100":因為新欄位是 NOT NULL,Alembic 自動幫既有資料填上預設值 100,避免升級失敗。如果忘了加這個預設值,既有資料行的 power_level 會是 NULL,違反 NOT NULL 限制,SQLite 會拒絕執行遷移。這是 autogenerate 聰明的地方之一。
再執行升級:
alembic upgrade head
# 輸出範例:
# INFO [alembic.runtime.migration] Running upgrade abc123 -> def456, add hero power_level
驗證既有資料還在、新欄位也加上了:
# verify_v2.py
from sqlmodel import Session, create_engine, select
from app.models import Hero
engine = create_engine("sqlite:///./app.db")
with Session(engine) as session:
heroes = session.exec(select(Hero)).all()
for h in heroes:
print(f"{h.name} 戰力:{h.power_level}")
# 輸出:Spider-Man 戰力:100
既有英雄的戰力自動補上 100,新加的英雄也可以帶自訂值。這就是遷移管理最大的價值:變更可控、可審查、舊資料不丟失。
降級:出問題時回到上一版
升級跑得起來,降級也要能跑。實務上偶爾會遇到「上了正式環境才發現新欄位設計錯了」這類情境,必須快速回退:
# 退一版
alembic downgrade -1
# 輸出範例:
# INFO [alembic.runtime.migration] Running downgrade def456 -> abc123, add hero power_level
# 退到指定版本(用識別碼前幾碼)
alembic downgrade abc1
# 退到最初(移除所有表,請小心)
alembic downgrade base
-1 是「上一版」的縮寫,等同於 downgrade -1。實務上建議每次升級前先確認「上一版的 downgrade 也跑得動」,因為有些破壞性變更(刪欄位、改型別)一旦套用就很難無損回退。在 CI 流程裡加入「升級後立刻降級」的煙霧測試,是大型團隊常見的做法。
常見錯誤與踩雷
第一個常見的踩雷是「env.py 沒 import 模型」。如果你寫了 alembic revision --autogenerate,但結果是空的(什麼動作都沒有),九成是因為 env.py 沒 import 你的 SQLModel 模組。Alembic 沒有讀到 metadata,自然比對不出差異。記得 from app.models import Hero, Team 之類的 import 一定要有,否則自動偵測會完全失靈。
第二個常見問題是「SQLite 不支援所有 ALTER 指令」。SQLite 在某些操作上比 PostgreSQL 嚴格:例如改欄位型別時,SQLite 不支援直接 ALTER COLUMN ... TYPE ...,必須用「建新表、複製資料、刪舊表、改名新表」這種招數。Alembic 對 SQLite 有內建支援(會自動用 batch mode),但偶爾會遇到 autogenerate 產生的腳本 SQLite 跑不動。遇到時,要把 upgrade() 改成 with op.batch_alter_table(...): 區塊,明確告訴 Alembic 用批次方式改表。
第三個是「忘記 downgrade 對稱」。很多新手寫了 upgrade 卻沒寫 downgrade,或 downgrade 寫錯(例如升級時 add_column,降級卻寫成 drop_constraint)。Alembic 不會自動幫你對稱,必須自己手動檢查每支遷移檔的升級與降級是否對稱。寫完一支遷移之後,至少要跑一次 alembic upgrade head && alembic downgrade -1 && alembic upgrade head,確認來回都沒問題。
第四個是「模型改了但忘記產生遷移」。本地端重新啟動應用程式時,如果還在用 SQLModel.metadata.create_all,SQLite 檔案會被悄悄加上新欄位,跟 Alembic 的狀態脫節。建議:一旦導入 Alembic,就把 create_all 從啟動流程移除,所有結構變更都走遷移。
效能與實務提醒
遷移管理的「正確性」比「效能」重要。在小專案裡,一支遷移檔可能只是「加一個欄位」,但在大型系統裡,遷移要套用到上千萬筆資料的表,每個 ALTER TABLE 都可能鎖表好幾分鐘。實務上有幾個原則:
第一,「不要在遷移裡跑大量資料更新」。如果必須搬資料(例如把舊欄位的值拆到新欄位),分兩階段:第一階段先加新欄位(NOT NULL DEFAULT 簡單值),部署後背景跑一支腳本把資料補齊,再發第二支遷移移除預設值。這樣每一支遷移都是「結構變更」,跑得快也不鎖表太久。
第二,「永遠把 alembic.ini 的 sqlalchemy.url 留空」。我們今天用環境變數注入網址,這樣 alembic.ini 不用改、也不會把密碼簽進版控。多環境部署時,建議用同一支遷移檔 + 不同環境變數。
第三,「autogenerate 不是萬能的」。它能偵測「新增/刪除欄位、新增/刪除表、新增/刪除索引」,但「改欄位型別」「重新命名欄位」很容易被誤判成「刪一個欄位加一個欄位」,導致資料丟失。遇到 rename 時,請改用手寫遷移,明確用 op.alter_column(... new_column_name=...)。
第四,「把遷移檔當成產品程式碼審查」。每支遷移檔都應該進 PR、經過同事審查,特別是 downgrade 區塊。資料庫變更不像程式碼可以隨時改,一支錯的遷移在正式環境跑掉,要花很大的力氣救回。
小結
今天我們把 SQLModel 的「結構變更」從手工作業升級成 Alembic 遷移流程。Alembic 用一條版本鏈追蹤所有變更,每次改模型就產生一支遷移檔,能升級也能降級;autogenerate 在簡單場景下非常方便,但對改欄位型別、rename、資料搬遷等情境還是要手寫。我們用 Hero 與 Team 兩張表示範了「第一次建表」「新增欄位」「降級」三種典型流程,並驗證了既有資料在升級後仍然完整。今天學的內容會在後續貫穿專案(Web Day 35 之後)持續用到,每一次部署都會跑一次 alembic upgrade head。
結語
今天解決了「資料庫結構」的版本管理問題。明天,我們要處理「執行流程」的另一個常見地雷:錯誤處理。FastAPI 雖然提供 HTTPException,但我們到目前為止只在「找資料找不到」時丟出 404;真實世界的 API 要處理驗證錯誤、權限錯誤、外部服務錯誤、伺服器內部錯誤等多種情境,每種都要回傳一致的 JSON 結構與對應的 HTTP 狀態碼。我們會設計一套「統一例外 + 自訂錯誤類別 + 全域處理器」的架構,讓前端永遠拿到結構相同的錯誤回應,並且讓開發者能快速定位問題。
延伸資源
- Alembic 官方文件(1.16,2025):
https://alembic.sqlalchemy.org/en/latest/ - SQLAlchemy 文件:遷移基礎(2025):
https://docs.sqlalchemy.org/en/20/core/metadata.html - SQLModel 文件:與 Alembic 整合(2025):
https://sqlmodel.tiangolo.com/advanced/decimal/ - FastAPI 官方教學:SQLModel + Alembic(2025):
https://fastapi.tiangolo.com/tutorial/sql-databases/
留言
張貼留言