快轉到主要內容
  1. 教學文章/

Python Alembic 實戰:資料庫 Migration、版本控管與團隊協作

·9 分鐘· loading · loading · ·
Python Alembic SQLAlchemy Database Migration Developer-Tools
每日拍拍
作者
每日拍拍
科學家 X 科技宅宅
目錄
Python 學習 - 本文屬於一個選集。
§ 68: 本文

featured

一. 前言:資料庫不是每次都能重建
#

嗨,這裡是拍拍君。寫 Python 專案時,資料表一開始通常很單純:一張 users、一張 orders、幾個欄位,看起來都合理。然後需求開始長大:使用者要多一個 display_name,訂單狀態要從字串改成 enum,某個欄位要加 index,舊資料還要補預設值。 這時候如果你還在靠 Base.metadata.create_all(),事情就有點危險了。create_all() 適合建立全新的資料庫,但它不會管理「已經存在的資料庫要怎麼安全演進」。正式資料庫裡有資料、有版本、有部署順序,不是刪掉重來就好。 這就是 Alembic 的位置。上一篇拍拍君寫過 SQLAlchemy 2.0 實戰,重點放在 typed ORM、Session 和查詢模式。今天這篇是更實務的後續:你的 ORM model 改了,資料庫要怎麼跟著改? 先講結論:

ORM model 是現在的樣子,migration 是資料庫從過去走到現在的路徑。 只保留 model,不保留路徑,正式環境會很痛。


二. 安裝與範例專案
#

這篇假設你已經知道 SQLAlchemy 的基本概念。如果還不熟,可以先看前一篇 SQLAlchemy 2.0 實戰。 先建立專案:

mkdir alembic-demo
cd alembic-demo
uv init
uv add sqlalchemy alembic
mkdir bookmark_app
touch bookmark_app/__init__.py
touch bookmark_app/db.py bookmark_app/models.py

如果你不用 uv,也可以用一般 venv:

python -m venv .venv
source .venv/bin/activate
python -m pip install sqlalchemy alembic

先準備最小的 SQLAlchemy 設定。

# bookmark_app/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker
class Base(DeclarativeBase):
    pass
DATABASE_URL = "sqlite:///bookmarks.db"
engine = create_engine(DATABASE_URL, echo=False)
SessionLocal = sessionmaker(bind=engine)

再放一個簡單的 model。

# bookmark_app/models.py
from sqlalchemy import String
from sqlalchemy.orm import Mapped, mapped_column
from bookmark_app.db import Base
class Bookmark(Base):
    __tablename__ = "bookmarks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    url: Mapped[str] = mapped_column(String(500), unique=True)

現在重點不是 ORM 多漂亮,而是要讓這張表的變化可以被版本控管。


三. 初始化 Alembic
#

在專案根目錄執行:

alembic init migrations

它會產生這些檔案:

alembic-demo/
├── alembic.ini
├── bookmark_app/
└── migrations/
    ├── env.py
    ├── script.py.mako
    └── versions/

幾個角色先認識一下:

  • alembic.ini:Alembic 的主要設定檔
  • migrations/env.py:每次 migration 執行時會載入的環境設定
  • migrations/versions/:每一個 migration script 放在這裡
  • script.py.mako:新 migration 檔案的模板 先不要急著改資料庫。Alembic 初始化後,最重要的是把它接到你的 SQLAlchemy 專案。

四. 設定資料庫 URL
#

打開 alembic.ini,會看到:

sqlalchemy.url = driver://user:pass@localhost/dbname

教學版本先改成:

sqlalchemy.url = sqlite:///bookmarks.db

這樣 Alembic 會連到同一個 SQLite 檔案。不過正式專案裡,拍拍君通常不建議把 production URL 寫死在 alembic.ini。比較常見的做法是從環境變數或 app settings 讀取,後面會講。


五. 讓 env.py 看懂你的 Model
#

Alembic 的 autogenerate 需要知道「目前 ORM model 長什麼樣子」。它靠的是 SQLAlchemy metadata。 打開 migrations/env.py,找到:

target_metadata = None

改成:

from bookmark_app.db import Base
from bookmark_app import models  # noqa: F401
target_metadata = Base.metadata

這裡的 models import 很重要。如果沒有 import model class,Base.metadata 可能是空的。Python 不會自動掃描你的資料夾;沒有載入 class,就沒有 table metadata。 這一步完成後,Alembic 才能比較資料庫現在有哪些 table、column、index,SQLAlchemy model 目前宣告了什麼,以及兩者差異可能需要生成哪些 migration。


六. 建立第一個 Migration
#

現在產生第一個 migration:

alembic revision --autogenerate -m "create bookmarks table"

你會在 migrations/versions/ 看到一個新檔案。打開它,核心大概像這樣:

from alembic import op
import sqlalchemy as sa
revision = "8c1f2a7d91b0"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
    op.create_table(
        "bookmarks",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sa.String(length=120), nullable=False),
        sa.Column("url", sa.String(length=500), nullable=False),
        sa.PrimaryKeyConstraint("id"),
        sa.UniqueConstraint("url"),
    )
def downgrade() -> None:
    op.drop_table("bookmarks")

兩個函式最重要:upgrade() 是往前升版,downgrade() 是往後回退。拍拍君建議你每次產生 migration 後都要打開看。autogenerate 是助手,不是審稿人。它可以抓到很多差異,但不能判斷你的商業邏輯。


七. 執行 Migration
#

把資料庫升到最新版:

alembic upgrade head

檢查目前版本:

alembic current

看歷史:

alembic history

這時候 SQLite 會多出兩個東西:bookmarks 是我們建立的資料表,alembic_version 是 Alembic 用來記錄目前資料庫版本的表。 alembic_version 很重要。Alembic 不是每次都重新掃描全部 migration 來猜狀態。它會看這張表,知道資料庫目前停在哪個 revision。 你可以用 sqlite CLI 看:

sqlite3 bookmarks.db ".tables"
sqlite3 bookmarks.db "select * from alembic_version;"

如果你看到 revision id,代表這個資料庫已經被 Alembic 管起來了。


八. 改 Model:新增欄位
#

現在需求來了。書籤需要 notes 欄位,讓使用者補一點說明。

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column
from bookmark_app.db import Base
class Bookmark(Base):
    __tablename__ = "bookmarks"
    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(120))
    url: Mapped[str] = mapped_column(String(500), unique=True)
    notes: Mapped[str | None] = mapped_column(Text, nullable=True)

再產生 migration:

alembic revision --autogenerate -m "add notes to bookmarks"

自動產生的 upgrade() 可能像這樣:

def upgrade() -> None:
    op.add_column("bookmarks", sa.Column("notes", sa.Text(), nullable=True))
def downgrade() -> None:
    op.drop_column("bookmarks", "notes")

這種單純新增 nullable 欄位,autogenerate 通常很準。檢查無誤後執行:

alembic upgrade head

到這裡你已經有基本 migration workflow:

  1. 改 model
  2. 產生 migration
  3. 打開檢查
  4. 執行 migration
  5. 把 model 和 migration 一起 commit

九. 不要盲信 autogenerate
#

Alembic autogenerate 很好用,但有幾件事它不會完全理解。 第一種是欄位重新命名。假設你把 title 改成 name,Alembic 可能會判斷成新增 name、刪除 title。但你真正想要的是 rename:

op.alter_column("bookmarks", "title", new_column_name="name")

如果你照著 add + drop 跑,舊資料可能就沒了。這不是 Alembic 壞。它只看到「舊欄位不見、新欄位出現」,不知道你腦袋裡其實是改名。 第二種是複雜資料轉換。例如你把 status 從自由字串改成固定 enum,Alembic 不知道舊資料要怎麼對應,你必須自己寫資料清理邏輯。 第三種是資料庫特定功能。PostgreSQL enum、partial index、trigger、extension,常常需要手寫 migration。 所以規則很簡單:

autogenerate 產生草稿,工程師負責判斷。


十. 資料遷移:expand、backfill、constrain
#

有些 migration 不只改欄位,也要改資料。例如我們希望舊資料的 created_at 不要是 NULL。比較穩的流程不是一開始就加 non-null,而是先允許 NULL、補資料、最後再加限制。

from datetime import datetime, timezone
from alembic import op
import sqlalchemy as sa
bookmark_table = sa.table(
    "bookmarks",
    sa.column("created_at", sa.DateTime()),
)
def upgrade() -> None:
    op.add_column(
        "bookmarks",
        sa.Column("created_at", sa.DateTime(), nullable=True),
    )
    op.execute(
        bookmark_table.update()
        .where(bookmark_table.c.created_at.is_(None))
        .values(created_at=datetime.now(timezone.utc).replace(tzinfo=None))
    )
    op.alter_column("bookmarks", "created_at", nullable=False)

這段做了三件事:先加 nullable 欄位,幫舊資料補值,再改成 non-null。這就是 schema migration 裡很常見的節奏:

expand -> backfill -> constrain

正式環境的資料庫尤其需要這樣想。你不是只在改 schema,你是在帶著既有資料一起過橋。


十一. 常用指令速查
#

日常最常用大概是這些:

alembic revision -m "describe change"
alembic revision --autogenerate -m "describe change"
alembic upgrade head
alembic upgrade <revision>
alembic downgrade <revision>
alembic downgrade -1
alembic current
alembic history
alembic heads
alembic upgrade head --sql

其中 head 代表目前 migration graph 的最前端。大多數小專案只有一條線,所以只有一個 head。多人同時新增 migration 時,可能會出現多個 head,下一節會講。


十二. 團隊協作:Migration 衝突怎麼辦
#

兩個人同時從同一版開發,可能會產生這種狀態:

base
 ├── add user avatar
 └── add bookmark tags

兩個 migration 的 down_revision 都指向同一個舊 revision。合併到 main 後,Alembic 會看到兩個 heads。 先檢查:

alembic heads

如果真的有多個 head,可以建立 merge migration:

alembic merge heads -m "merge migration heads"

它會產生一個 migration,down_revision 指向兩個 parent。內容可能是空的:

def upgrade() -> None:
    pass
def downgrade() -> None:
    pass

這不是偷懶。它的目的只是把 migration graph 合成一條共同路徑。不過,如果兩個 migration 改到同一張表、同一個欄位附近,還是要人工檢查順序和資料影響。merge heads 解的是圖的形狀,不是業務衝突。


十三. 設定不要寫死
#

前面為了教學,把 URL 寫在 alembic.ini。正式專案通常會改成從環境變數讀。 migrations/env.py 可以這樣處理:

import os
from alembic import context
config = context.config
database_url = os.environ.get("DATABASE_URL")
if database_url:
    config.set_main_option("sqlalchemy.url", database_url)

然後執行時:

DATABASE_URL=sqlite:///bookmarks.db alembic upgrade head

如果你已經有 settings 模組,也可以從那裡讀。重點是不要讓 production 密碼躺在 git 裡。這跟 pydantic-settings 的精神很像:設定可以被版本控管,秘密不應該被版本控管。


十四. FastAPI 專案裡怎麼放
#

如果你的專案是 FastAPI,常見結構會是 alembic.ini、migrations/、app/db.py、app/models/ 和 app/settings.py 分開放。migrations/env.py 裡要確保所有 model 都被 import;一個常見做法是在 app/models/init.py 集中 import User、Bookmark 等 model,然後 env.py import app.models,讓 Base.metadata 看得到所有 table。 不要在 FastAPI startup 裡偷偷 create_all()。開發初期看似方便,正式環境會讓 migration 狀態變模糊。比較乾淨的規則是:app 啟動只連資料庫,migration 由部署流程或維運指令執行,model 變更一定要附 migration。


十五. 測試 Migration
#

Migration 也可以測。尤其是有資料轉換的 migration,最好不要只靠肉眼看。 最基本的 smoke test 可以是:

rm -f test.db
DATABASE_URL=sqlite:///test.db alembic upgrade head
DATABASE_URL=sqlite:///test.db alembic downgrade base

這能確認全部 migration 可以從空資料庫跑到最新版,也能確認 downgrade 至少沒有語法錯。 如果你要放進 pytest,可以用 subprocess 跑 Alembic,把 DATABASE_URL 指到 tmp_path 裡的測試資料庫,再執行 upgrade head。測 migration 的目的不是追求覆蓋率漂亮,而是提早發現「這個 migration 在乾淨環境根本跑不起來」或「舊資料轉換會爆」。


十六. 常見坑
#

第一個坑:忘記 import model。結果 autogenerate 以為所有 table 都被刪掉,產生一堆 drop_table()。看到這種 migration,請先停手,不要勇敢。 第二個坑:nullable 欄位改 non-null。正式資料庫有舊資料時,要先 backfill。 第三個坑:把 migration 當成可隨便重寫的草稿。如果 migration 已經進 main,甚至已經跑到 staging 或 production,就不要隨便改舊檔。新增一個修正 migration 通常比較安全。 第四個坑:不同環境 migration 狀態不一致。請用 alembic current 檢查,不要用「我記得應該跑過」管理 production。記憶力不是部署工具。 第五個坑:SQLite 和 PostgreSQL 行為不同。SQLite 很適合教學和本機測試,但不是所有 ALTER TABLE 都一樣。如果 production 是 PostgreSQL,CI 最好也用 PostgreSQL 跑 migration 測試。


十七. 拍拍君的實務規則
#

拍拍君通常會用這幾條規則管 Alembic:

  • 每次 model schema 變更都要有 migration
  • migration 檔案要進 code review
  • autogenerate 後一定手動檢查
  • 重新命名欄位要特別小心,不接受盲目的 add/drop
  • 牽涉舊資料時,使用 expand -> backfill -> constrain
  • production migration 前先在 staging 跑過
  • 不把 production database URL 寫進 git
  • 已經被共享環境使用的 migration 不任意改寫 這些規則聽起來有點囉嗦。但資料庫壞掉時,囉嗦會突然變得很可愛。

十八. 結語
#

Alembic 最重要的價值,不是讓你少打幾行 SQL。它真正解決的是「資料庫變更也要有版本歷史」。 SQLAlchemy model 告訴你現在的資料結構。Alembic migration 告訴你資料庫怎麼一步一步走到現在。在個人小專案裡,這會讓你比較敢改 schema。在團隊專案裡,這會讓 code review、部署和 rollback 都有根據。 今天先記住這條線就夠了:

改 model -> 產生 migration -> 檢查 migration -> 測試 -> upgrade

不要把資料庫當成可以隨手改的黑箱。它會記得你做過什麼。你也應該用 migration 記得。


延伸閱讀
#

Python 學習 - 本文屬於一個選集。
§ 68: 本文

相關文章

Python SQLAlchemy 2.0 實戰:Typed ORM、Session 與查詢模式
·9 分鐘· loading · loading
Python SQLAlchemy ORM Database SQLite Developer-Tools
Python pytest fixtures 進階:conftest、factory 與測試資料管理
·8 分鐘· loading · loading
Python Pytest Fixtures Testing Conftest Monkeypatch Developer-Tools
FastAPI + Streamlit 實戰:API 後端與互動前端分工
·9 分鐘· loading · loading
Python Fastapi Streamlit Api Frontend Developer-Tools
Python pydantic-settings 實戰:型別安全管理 .env 與設定檔
·6 分鐘· loading · loading
Python Pydantic Pydantic-Settings Dotenv Configuration Developer-Tools
Python tempfile 實戰:安全建立暫存檔案、目錄與測試資料
·9 分鐘· loading · loading
Python Tempfile Filesystem Testing Standard-Library Developer-Tools
Python difflib 實戰:文字差異比對、相似度比較與 patch 輸出完全攻略
·10 分鐘· loading · loading
Python Difflib Text-Processing Developer-Tools Cli