# 

# 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    # 新增欄位
```

---

## 📦 安裝與初始化

### 安裝

```bash
pip install alembic
```

### 初始化

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

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

---

## 🔧 設定 Alembic

### 1. 修改 alembic.ini

```ini
# alembic.ini

[alembic]
script_location = alembic
prepend_sys_path = .

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

### 2. 修改 env.py

```python
# 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

```python
# 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()
```

---

## 📝 建立遷移

### 自動生成遷移腳本

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

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

### 手動建立遷移腳本

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

### 遷移腳本範例

```python
# 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')
```

---

## 🚀 執行遷移

### 常用指令

```bash
# 升級到最新版本
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

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

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

---

## 🔧 常見遷移操作

### 新增欄位

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

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

### 刪除欄位

```python
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))
```

### 修改欄位

```python
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
    )
```

### 重新命名欄位

```python
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')
```

### 新增索引

```python
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')
```

### 新增外鍵

```python
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')
```

### 建立新表

```python
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')
```

### 資料遷移

```python
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. 自動生成未偵測到變更

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

### 2. 處理既有資料

```python
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. 分支合併

```bash
# 查看分支
alembic heads

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

### 4. 錯誤回滾

```bash
# 如果遷移執行失敗，手動標記版本
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
└── ...
```

### 命名規範

```bash
# 遷移訊息格式
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"
```

### 遷移腳本檢查清單

```python
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 整合

```yaml
# 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）讓我們可以：
>
> 1. **版本控制**：追蹤 Schema 變更歷史
> 2. **環境同步**：開發、測試、生產環境保持一致
> 3. **團隊協作**：多人開發時不會衝突
> 4. **安全回滾**：出問題可以降級
>
> 使用方式：
> ```bash
> # 自動偵測 Model 變更
> alembic revision --autogenerate -m "add user table"
>
> # 執行遷移
> alembic upgrade head
> ```

---

**上一篇：** [03-3. 非同步資料庫操作](./03-3)
**下一篇：** [03-5. Repository 模式](./03-5)

---

最後更新：2025-12-17


---

> 作者: luk  
> URL: https://yoru-karu-blog-lalaluk-52581ac5e0cef170a3c8922c19182ecb6f7bd604.gitlab.io/posts/tutorial/fastapi/03-4/