数据库迁移是最容易出事故的工程操作之一。 Claude Code 能帮你生成迁移脚本、设计安全策略，避免那些可能导致停机的坑。
为什么数据库迁移需要 AI 辅助？
迁移的难点：正确的迁移脚本（索引/约束不遗漏）、生产环境安全（大表迁移可能锁表）、回滚策略、历史数据迁移（格式转换/空值处理）。
Alembic（Python/SQLAlchemy 项目）
基本工作流
bash
# 初始化
alembic init alembic
# 自动检测 Model 变更生成迁移
alembic revision --autogenerate -m 'add_user_phone_column'
# 执行迁移
alembic upgrade head
# 回滚一步
alembic downgrade -1安全添加列（生产 50 万数据）
Prompt：帮我给 User 表添加 phone_number 字段（String(20)，可空），添加允许多个 NULL 的唯一索引，生产库有 50 万数据，说明是否会锁表。
Claude Code 生成：
python
# alembic/versions/20260401_add_user_phone.py
from alembic import op
import sqlalchemy as sa
revision = '20260401_001'
down_revision = '20260331_005'
def upgrade():
# 1. 添加可空列（不锁表，快速完成）
op.add_column('users',
sa.Column('phone_number', sa.String(20), nullable=True)
)
# 2. PostgreSQL：CONCURRENTLY 建索引不锁表
op.execute(
'CREATE UNIQUE INDEX CONCURRENTLY idx_users_phone '
'ON users(phone_number) WHERE phone_number IS NOT NULL'
)
def downgrade():
op.drop_index('idx_users_phone', table_name='users')
op.drop_column('users', 'phone_number')Claude Code 同时给出风险分析：
- PostgreSQL：CONCURRENTLY 建索引不锁表，安全
- MySQL 8.0+：Online DDL，50 万行约 1-3 分钟
- MySQL 5.6 以下：会锁表，建议使用 pt-online-schema-change
生产环境零停机：先加后删原则
场景：把 full_name 列拆分为 first_name + last_name，200 万数据，不停机。
Claude Code 设计的 4 步方案：
第 1 步（部署 v1）：添加新列，保留旧列
应用写入时同时写新旧两列
第 2 步（数据迁移）：分批回填历史数据
每批 1000 条，避免长事务
第 3 步（部署 v2）：切换到只读新列
full_name 改为由应用层拼接
第 4 步（部署 v3，一周后）：删除旧列
DROP COLUMN full_name
Flyway（Java/Spring Boot）
目录结构
src/main/resources/db/migration/
V1__create_users_table.sql
V2__add_email_index.sql
V3__add_phone_column.sql
Prompt：给 orders 表（500 万行）添加联合索引 (user_id, status, created_at DESC)，生成 Flyway V8 脚本。
Claude Code 生成：
sql
-- V8__add_orders_user_status_index.sql
-- 注意：500万行表，建索引预计耗时 3-10 分钟
-- MySQL 8.0+ Online DDL：不锁读，写操作短暂等待
-- 建议在低峰期执行，或使用 pt-online-schema-change
ALTER TABLE orders
ADD INDEX idx_user_status (user_id, status, created_at DESC),
ALGORITHM=INPLACE, LOCK=NONE;数据格式迁移脚本
Prompt：users 表的 preferences 列是旧格式 JSON，需要迁移到新格式，分批处理（每批 500 条），失败可重试，记录进度方便中断后继续。
python
import json, time
from sqlalchemy import text
def migrate_preferences(db, batch_size=500):
offset = 0
while True:
rows = db.execute(
text('SELECT id, preferences FROM users '
'WHERE migrated_at IS NULL LIMIT :limit OFFSET :offset'),
{'limit': batch_size, 'offset': offset}
).fetchall()
if not rows:
break
for row in rows:
old = json.loads(row.preferences)
new_prefs = {
'ui': {'theme': old.get('theme', 'light')},
'locale': 'zh-CN' if old.get('lang') == 'zh' else 'en-US',
'notify': {'enabled': old.get('notifications', True)}
}
db.execute(
text('UPDATE users SET preferences=:p, migrated_at=NOW() WHERE id=:id'),
{'p': json.dumps(new_prefs, ensure_ascii=False), 'id': row.id}
)
db.commit()
print(f'批次完成：offset={offset}, 处理 {len(rows)} 条')
offset += batch_size
time.sleep(0.1) # 避免对数据库施加过大压力回滚 SOP
1. 立即评估（< 5 分钟）
是结构变更还是数据变更？是否有数据已用新格式写入？
2. 结构变更回滚
alembic downgrade -1
3. 数据变更回滚
有备份则恢复；无备份则执行反向 UPDATE 脚本
4. 应用同步回滚到上一版本
5. 事后复盘：为什么 staging 没发现？迁移前是否备份？
来源：Alembic 官方文档 + Flyway 官方文档