luk

0001-01-01 約 1100 字預計閱讀 6 分鐘

03-4. 資料庫遷移 Alembic

⏱️ 閱讀時間： 20 分鐘 🎯 難度： ⭐⭐⭐ (進階)

🤔 一句話解釋

Alembic 是 SQLAlchemy 的資料庫遷移工具，讓你用版本控制的方式管理資料庫 Schema 變更。

🎯 為什麼需要資料庫遷移？

沒有遷移工具的問題

❌ 直接修改資料庫：
- 無法追蹤變更歷史
- 無法在不同環境同步 Schema
- 回滾困難
- 團隊協作混亂

✅ 使用 Alembic：
- 版本控制 Schema 變更
- 自動生成遷移腳本
- 支援升級和降級
- 團隊協作順暢

遷移流程

修改 Model  →  生成遷移腳本  →  審核腳本  →  執行遷移
    ↓              ↓              ↓           ↓
class User    alembic revision   檢查 SQL   alembic upgrade
  id: int      --autogenerate
  name: str
  + email: str    # 新增欄位

📦 安裝與初始化

安裝

pip install alembic

初始化

# 在專案根目錄執行
alembic init alembic

# 會產生以下結構：
# alembic/
# ├── env.py           # 遷移環境設定
# ├── README
# ├── script.py.mako   # 遷移腳本模板
# └── versions/        # 遷移腳本存放處
# alembic.ini          # Alembic 設定檔

🔧 設定 Alembic

1. 修改 alembic.ini

# alembic.ini

[alembic]
script_location = alembic
prepend_sys_path = .

# 資料庫 URL（可以在 env.py 中覆蓋）
sqlalchemy.url = postgresql://user:password@localhost:5432/mydb

2. 修改 env.py

# alembic/env.py

from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
import os
import sys

# 加入專案路徑
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 匯入你的 Model
from app.models import Base  # 根據你的專案結構調整
from app.config import settings  # 從設定檔讀取資料庫 URL

config = context.config

# 從環境變數或設定檔讀取資料庫 URL
config.set_main_option("sqlalchemy.url", settings.database_url)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# 設定 metadata（重要！）
target_metadata = Base.metadata


def run_migrations_offline() -> None:
    """離線模式執行遷移"""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """線上模式執行遷移"""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True,  # 比較欄位型別變更
            compare_server_default=True,  # 比較預設值變更
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

3. 非同步版本的 env.py

# alembic/env.py（非同步版本）

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.models import Base
from app.config import settings

config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def run_migrations_offline() -> None:
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def do_run_migrations(connection: Connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,
        compare_server_default=True,
    )

    with context.begin_transaction():
        context.run_migrations()


async def run_async_migrations() -> None:
    """非同步執行遷移"""
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()


def run_migrations_online() -> None:
    asyncio.run(run_async_migrations())


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

📝 建立遷移

自動生成遷移腳本

# 根據 Model 變更自動生成遷移
alembic revision --autogenerate -m "add user table"

# 會在 alembic/versions/ 產生類似：
# xxxxxxxxxxxx_add_user_table.py

手動建立遷移腳本

# 建立空白遷移腳本
alembic revision -m "custom migration"

遷移腳本範例

# alembic/versions/xxxx_add_user_table.py

"""add user table

Revision ID: abc123
Revises:
Create Date: 2025-12-17 10:00:00.000000

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision: str = 'abc123'
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(
        'users',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sa.String(length=50), nullable=False),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('hashed_password', sa.String(length=255), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False, server_default='true'),
        sa.Column('created_at', sa.DateTime(), nullable=False, server_default=sa.func.now()),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('ix_users_username', 'users', ['username'], unique=True)
    op.create_index('ix_users_email', 'users', ['email'], unique=True)


def downgrade() -> None:
    """降級：回滾變更"""
    op.drop_index('ix_users_email', table_name='users')
    op.drop_index('ix_users_username', table_name='users')
    op.drop_table('users')

🚀 執行遷移

常用指令

# 升級到最新版本
alembic upgrade head

# 升級到特定版本
alembic upgrade abc123

# 升級一個版本
alembic upgrade +1

# 降級一個版本
alembic downgrade -1

# 降級到特定版本
alembic downgrade abc123

# 降級到最初狀態
alembic downgrade base

# 查看當前版本
alembic current

# 查看遷移歷史
alembic history

# 查看詳細歷史
alembic history --verbose

查看將要執行的 SQL

# 查看升級 SQL（不實際執行）
alembic upgrade head --sql

# 儲存 SQL 到檔案
alembic upgrade head --sql > migration.sql

🔧 常見遷移操作

新增欄位

def upgrade() -> None:
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

def downgrade() -> None:
    op.drop_column('users', 'phone')

刪除欄位

def upgrade() -> None:
    op.drop_column('users', 'old_field')

def downgrade() -> None:
    op.add_column('users', sa.Column('old_field', sa.String(100), nullable=True))

修改欄位

def upgrade() -> None:
    # 修改欄位型別
    op.alter_column(
        'users',
        'phone',
        existing_type=sa.String(20),
        type_=sa.String(50),
        existing_nullable=True
    )

    # 修改為非空
    op.alter_column(
        'users',
        'email',
        existing_type=sa.String(100),
        nullable=False
    )

def downgrade() -> None:
    op.alter_column(
        'users',
        'phone',
        existing_type=sa.String(50),
        type_=sa.String(20),
        existing_nullable=True
    )

    op.alter_column(
        'users',
        'email',
        existing_type=sa.String(100),
        nullable=True
    )

重新命名欄位

def upgrade() -> None:
    op.alter_column('users', 'name', new_column_name='full_name')

def downgrade() -> None:
    op.alter_column('users', 'full_name', new_column_name='name')

新增索引

def upgrade() -> None:
    op.create_index('ix_users_created_at', 'users', ['created_at'])

def downgrade() -> None:
    op.drop_index('ix_users_created_at', table_name='users')

新增外鍵

def upgrade() -> None:
    op.add_column('posts', sa.Column('author_id', sa.Integer(), nullable=False))
    op.create_foreign_key(
        'fk_posts_author_id',
        'posts',
        'users',
        ['author_id'],
        ['id'],
        ondelete='CASCADE'
    )

def downgrade() -> None:
    op.drop_constraint('fk_posts_author_id', 'posts', type_='foreignkey')
    op.drop_column('posts', 'author_id')

建立新表

def upgrade() -> None:
    op.create_table(
        'categories',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(50), nullable=False),
        sa.Column('slug', sa.String(50), nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.UniqueConstraint('slug')
    )

def downgrade() -> None:
    op.drop_table('categories')

資料遷移

from sqlalchemy import table, column, String

def upgrade() -> None:
    # 1. 新增欄位
    op.add_column('users', sa.Column('status', sa.String(20), nullable=True))

    # 2. 資料遷移
    users = table('users', column('status', String))
    op.execute(
        users.update().values(status='active')
    )

    # 3. 設為非空
    op.alter_column('users', 'status', nullable=False)

def downgrade() -> None:
    op.drop_column('users', 'status')

⚠️ 常見問題與解決

1. 自動生成未偵測到變更

# 在 env.py 中加入
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    compare_type=True,           # 偵測型別變更
    compare_server_default=True, # 偵測預設值變更
)

2. 處理既有資料

def upgrade() -> None:
    # 新增欄位時，先允許 NULL
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

    # 填入預設值
    op.execute("UPDATE users SET phone = '' WHERE phone IS NULL")

    # 再改為 NOT NULL
    op.alter_column('users', 'phone', nullable=False)

3. 分支合併

# 查看分支
alembic heads

# 如果有多個 head，需要合併
alembic merge -m "merge heads" abc123 def456

4. 錯誤回滾

# 如果遷移執行失敗，手動標記版本
alembic stamp abc123

# 然後修復遷移腳本，重新執行
alembic upgrade head

📝 最佳實踐

專案結構

project/
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
│       ├── 001_initial.py
│       ├── 002_add_user_email.py
│       └── 003_add_posts_table.py
├── alembic.ini
├── app/
│   ├── models/
│   │   ├── __init__.py
│   │   ├── base.py
│   │   └── user.py
│   └── config.py
└── ...

命名規範

# 遷移訊息格式
alembic revision --autogenerate -m "add_user_table"
alembic revision --autogenerate -m "add_email_to_users"
alembic revision --autogenerate -m "create_posts_table"
alembic revision --autogenerate -m "add_index_on_created_at"

遷移腳本檢查清單

def upgrade() -> None:
    # ✅ 檢查是否有資料
    # ✅ 考慮大表的效能影響
    # ✅ 使用批次處理大量更新
    # ✅ 先加欄位（nullable），再遷移資料，最後改 nullable

    # 範例：安全地新增非空欄位
    op.add_column('users', sa.Column('role', sa.String(20), nullable=True))
    op.execute("UPDATE users SET role = 'user' WHERE role IS NULL")
    op.alter_column('users', 'role', nullable=False)

def downgrade() -> None:
    # ✅ 確保可以回滾
    # ✅ 考慮資料損失的影響
    op.drop_column('users', 'role')

CI/CD 整合

# GitHub Actions 範例
name: Database Migration

on:
  push:
    branches: [main]

jobs:
  migrate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run migrations
        env:
          DATABASE_URL: ${{ secrets.DATABASE_URL }}
        run: alembic upgrade head

✅ 重點總結

常用指令

指令	說明
`alembic init alembic`	初始化
`alembic revision --autogenerate -m "msg"`	自動生成遷移
`alembic upgrade head`	升級到最新
`alembic downgrade -1`	降級一版
`alembic current`	查看當前版本
`alembic history`	查看歷史

遷移操作

操作	函數
新增表	`op.create_table()`
刪除表	`op.drop_table()`
新增欄位	`op.add_column()`
刪除欄位	`op.drop_column()`
修改欄位	`op.alter_column()`
新增索引	`op.create_index()`
新增外鍵	`op.create_foreign_key()`

🎤 面試這樣答

Q: 為什麼需要資料庫遷移工具？

答案：

資料庫遷移工具（如 Alembic）讓我們可以：
版本控制：追蹤 Schema 變更歷史
環境同步：開發、測試、生產環境保持一致
團隊協作：多人開發時不會衝突
安全回滾：出問題可以降級
使用方式：
# 自動偵測 Model 變更
alembic revision --autogenerate -m "add user table"

# 執行遷移
alembic upgrade head

上一篇： 03-3. 非同步資料庫操作 下一篇： 03-5. Repository 模式

最後更新：2025-12-17