需要为 SQLAlchemy/PostgreSQL…

适用于 Python + SQLAlchemy/PostgreSQL 项目，用 Alembic 创建、审查、测试、部署和回滚数据库迁移，保障多环境与生产环境数据一致性。

适用场景

解决思路

1. 按迁移生命周期操作：开发本地创建迁移，staging 验证，生产前 code review 迁移脚本，生产部署时尽量最小停机，上线后监控状态并处理失败，必要时回滚。
2. 安装依赖：`pip install alembic psycopg2-binary sqlalchemy`，或在 requirements.txt 加入 `alembic>=1.13.0`、`sqlalchemy>=2.0.0`、`psycopg2-binary>=2.9.0`。
3. 初始化 Alembic：运行 `alembic init alembic` 生成 `alembic/`、`alembic/versions/`、`alembic/env.py`、`alembic.ini`；多数据库用 `alembic init --template multidb alembic`。
4. 配置 `alembic.ini` 的 `sqlalchemy.url`：开发环境写 `postgresql://user:password@localhost/support_dev`；生产环境用 `postgresql://%(DB_USER)s:%(DB_PASSWORD)s@%(DB_HOST)s/%(DB_NAME)s`。
5. 在 `env.py` 用环境变量覆盖连接：读取 `DATABASE_URL`，默认 `postgresql://localhost/support_dev`，调用 `config.set_main_option('sqlalchemy.url', db_url)`，并设置 `target_metadata = Base.metadata`。
6. 手工创建迁移：`alembic revision -m 'add ticket priority column'`，在生成的 `versions/...py` 中设置 `revision`、`down_revision`、`branch_labels`、`depends_on`。
7. 在 `upgrade()` 中加列：`op.add_column('tickets', sa.Column('priority', sa.String(20), nullable=True, server_default='normal'))`；建索引：`op.create_index('ix_tickets_priority', 'tickets', ['priority'])`。
8. 在 `downgrade()` 中按逆序删索引：`op.drop_index('ix_tickets_priority', 'tickets')`；删列：`op.drop_column('tickets', 'priority')`。
9. 自动生成迁移：`alembic revision --autogenerate -m 'add customer satisfaction table'`；生成后必须人工审查，尤其检查重命名列、需数据转换的类型变更、复杂约束、数据迁移。
10. 自动生成建表示例要点：用 `op.create_table('customer_satisfaction', ...)`，定义 `sa.Column`、`sa.ForeignKeyConstraint(['ticket_id'], ['tickets.id'], ondelete='CASCADE')`、`sa.PrimaryKeyConstraint('id')`，并建索引 `op.create_index(...)`；`downgrade()` 逆序 drop index 和 drop_table。
11. 数据迁移：先 `op.add_column('tickets', sa.Column('status_new', sa.String(50), nullable=True))`，用 `table()`/`column()` 定义表结构，建立 `status_mapping`，通过 `connection = op.get_bind()` 执行 `tickets.update().where(...).values(...)` 批量更新。
12. 数据迁移继续：数据迁移完成后 `op.alter_column('tickets', 'status_new', nullable=False)`，再 `op.drop_column('tickets', 'status')`，最后 `op.alter_column('tickets', 'status_new', new_column_name='status')`。
13. 数据迁移回滚：在 `downgrade()` 建临时列，定义反向 `reverse_mapping`，用 `op.get_bind()` 执行反向 update，设非空，删当前列，再把临时列改回原名。
14. 大数据量分批迁移：先加可空列，用 `select(...).where(...).limit(batch_size).offset(offset)` 循环取批，`batch_size = 1000`，逐批 update，最后 `op.alter_column(..., nullable=False, server_default='0')`；`downgrade()` 删除该列。
15. 升级命令：`alembic upgrade head`；只生成 SQL 不执行：`alembic upgrade head --sql`；单步升级：`alembic upgrade +1`；升到指定版本：`alembic upgrade abc123`。
16. 降级命令：`alembic downgrade -1`；降到指定版本：`alembic downgrade abc123`；降到空库：`alembic downgrade base`；只生成降级 SQL：`alembic downgrade -1 --sql`。
17. 查看状态：`alembic current`、`alembic current --verbose`、`alembic history`、`alembic history --indicate-current`、`alembic history -r base:head`。
18. 创建分支：`alembic revision -m 'create reporting branch' --head=base --branch-label=reporting --version-path=alembic/versions/reporting`；向分支加迁移：`alembic revision -m 'add report tables' --head=reporting@head`。
19. 多分支操作：`alembic heads` 查看所有分支头，`alembic branches` 查看分支点，`alembic upgrade reporting@head` 升级指定分支，`alembic upgrade heads` 升级所有分支。
20. 合并分支：`alembic merge -m 'merge reporting into main' main@head reporting@head`；生成的 merge migration 中 `down_revision = ('ghi789', 'rep002')`，通常 `upgrade()`/`downgrade()` 为 `pass`，如有冲突再写协调逻辑。
21. 跨分支依赖：创建迁移时用 `--depends-on=def456` 指定依赖某个主分支 revision，例如 `alembic revision -m 'reporting needs user table' --head=reporting@head --depends-on=def456`。
22. 测试迁移：用 pytest fixture 构造 `Config('alembic.ini')`，`config.set_main_option('sqlalchemy.url', 'postgresql://localhost/support_test')`；测试库 fixture 调用 `command.upgrade(alembic_config, 'head')`，yield 后 `command.downgrade(alembic_config, 'base')` 并 `engine.dispose()`。
23. 测试表结构：用 `inspect(test_db).get_table_names()` 断言 `tickets`、`users`、`customer_satisfaction` 存在；用 `inspector.get_columns('tickets')` 断言列存在并检查 `columns['priority']['type'].python_type == str` 等类型。
24. 测试升级降级循环：依次执行 `command.downgrade(alembic_config, 'base')`、`command.upgrade(alembic_config, 'head')`、`command.downgrade(alembic_config, '-1')`、`command.upgrade(alembic_config, 'head')`，确认无错误。
25. 测试数据迁移不丢数据：插入一条 `Ticket`，提交并记录 `ticket_id`，执行相关迁移后查回该记录，断言 `retrieved.title == 'Test ticket'` 等字段。
26. 测试无待执行迁移：用 `ScriptDirectory.from_config(alembic_config)` 和 `MigrationContext.configure(connection)`，比较 `context.get_current_heads()` 与 `script.get_heads()`，必须相等。
27. 测试迁移顺序：用 `script.walk_revisions()` 遍历所有 revision，检查 `down_revision` 非空时对应的 revision 存在；若是 tuple（merge 点）则逐个检查。
28. 测试 schema drift：调用 `command.check(alembic_config)`，无异常说明数据库与模型一致；有异常则报告 drift。
29. 测试迁移性能：降级到 base 后记录 `command.upgrade(alembic_config, 'head')` 耗时，断言小于 60 秒；用 10000 条 Ticket 数据测试数据迁移，断言 10000 条在 30 秒内完成。
30. CI 工作流（GitHub Actions）：在 `pull_request` 与 `push` 到 `main`/`develop` 时触发，监听 `alembic/versions/**`、`myapp/models/**`、`alembic.ini`、`alembic/env.py`。
31. CI 服务：启动 `postgres:15` service，设置 `POSTGRES_PASSWORD: postgres`、`POSTGRES_DB: support_test`，健康检查 `pg_isready`，端口 `5432:5432`。
32. CI 步骤：`actions/checkout@v3`、`actions/setup-python@v4` 指定 Python 3.11、`pip install -r requirements.txt` 和 `pytest pytest-cov`。
33. CI 迁移测试：设置 `DATABASE_URL=postgresql://postgres:postgres@localhost/support_test`，依次运行 `alembic upgrade head`、`alembic downgrade base`、`alembic upgrade head`、`pytest tests/test_migrations.py -v`。
34. CI 检查 drift：运行 `alembic check`。
35. CI 校验迁移历史：统计 `alembic heads | wc -l`，若大于 1 则报错 `Multiple heads detected. Please merge branches.` 并执行 `alembic heads` 后退出 1。
36. CI 审查 SQL：在 PR 中运行 `alembic upgrade head --sql > migration.sql`，用 `actions/upload-artifact@v3` 上传 `migration.sql`，再用 `actions/github-script@v6` 把 SQL 评论到 PR。
37. 部署脚本：`set -e`；设置 `DB_HOST`、`DB_NAME`、`DB_USER`、`DATABASE_URL`；用 `pg_dump '$DATABASE_URL' > '$BACKUP_FILE'` 备份到 `./backups/pre_migration_${TIMESTAMP}.sql`。
38. 部署脚本继续：运行 `alembic current` 查看当前状态，`alembic history --verbose | grep -A 5 'head'` 查看待执行；用 `timeout 300 alembic upgrade head` 执行，失败或超时则 `psql '$DATABASE_URL' < '$BACKUP_FILE'` 恢复并退出。
39. 部署脚本验证：比较 `alembic current` 的 revision 与 `alembic heads` 的 head；不一致则从备份恢复并退出；成功后用 `ls -t '$BACKUP_DIR'/*.sql | tail -n +11 | xargs -r rm` 只保留最近 10 个备份。
40. 生产前检查清单：开发环境测试、staging 用生产相似数据测试、至少一人 review、测试 downgrade 路径、评估大表性能影响、准备数据库备份、记录回滚流程、必要时安排维护窗口、通知团队、配置监控告警。
41. 零停机迁移 Phase 1：新增字段时先设 `nullable=True`，例如 `op.add_column('tickets', sa.Column('new_field', sa.String(100), nullable=True))`，`downgrade()` 删列。
42. 零停机迁移 Phase 2：后台小批量回填数据，用 `connection.execute('UPDATE tickets SET new_field = calculate_value(old_field) WHERE new_field IS NULL LIMIT {batch_size}')`，`batch_size = 100`，每批后 `time.sleep(0.1)`，直到 `result.rowcount == 0`。
43. 零停机迁移 Phase 3：数据回填完成后，用 `op.alter_column('tickets', 'new_field', nullable=False, server_default='default_value')` 设为必填；回滚时恢复 `nullable=True, server_default=None`。
44. 零停机迁移 Phase 4：确认新字段稳定后，用 `op.drop_column('tickets', 'old_field')` 删旧列；回滚时重新添加旧列。
45. 迁移失败处理：在 `alembic/env.py` 的 `run_migrations_online()` 中配置 `transaction_per_migration=True`、`compare_type=True`、`compare_server_default=True`，用 try/except 捕获异常并记录日志，事务自动回滚后重新抛出。
46. 自定义迁移模板：修改 `alembic/script.py.mako`，在 docstring 中包含 `Author: ${author if author else 'Support Team'}`、`Jira: ${jira_ticket if jira_ticket else 'N/A'}`，并保留 `revision`、`down_revision`、`branch_labels`、`depends_on`。
47. 多数据库支持：在 `env.py` 中为每个数据库配置 `url` 与 `target_metadata`，例如 `main` 用 `MAIN_DB_URL` 和 `main_metadata`，`analytics` 用 `ANALYTICS_DB_URL` 和 `analytics_metadata`；循环中调用 `context.configure(..., upgrade_token=f'{name}_upgrade', downgrade_token=f'{name}_downgrade')`，并 `context.run_migrations(engine_name=name)`。
48. 多 head 错误处理：运行 `alembic merge heads -m 'merge branches'` 合并分支。
49. 迁移不同步处理：用 `alembic stamp head` 或 `alembic stamp abc123` 把数据库标记到指定 revision。
50. 失败迁移清理：先 `alembic current`，再进入 `psql $DATABASE_URL` 手工修复，再 `alembic stamp previous_working_revision`，最后 `alembic upgrade head`。
51. 循环依赖处理：创建迁移时用 `--depends-on=branch_b_revision` 而不是只依赖 `down_revision`，例如 `alembic revision -m 'fix circular dependency' --head=branch_a@head --depends-on=branch_b_revision`。
52. 收尾原则：始终 review autogenerated migrations；生产前彻底测试；重大迁移前备份；规划回滚；监控迁移性能；为复杂迁移写文档。

适用边界 · 注意事项

• 项目未使用 SQLAlchemy/Alembic，或数据库不是关系型（如 MongoDB），本技能不适用。
• 没有数据库备份、回滚预案，或无法承受 schema 锁表/停机风险的生产库，不要直接套用零停机/批量数据迁移方案。
• 仅需一次性手工改表，且不追求版本化、可复现、可回滚时，不必引入完整 Alembic 流程。