项目上线后，数据库的结构变更往往是风险最高的环节之一。无论是增加字段、调整索引还是创建新表，这些看似简单的操作在实际开发中常常引发问题：本地修改后忘记同步到测试环境；测试环境执行了脚本，生产环境却遗漏了关键的ALTER语句；团队协作时难以追踪哪些SQL已执行、哪些尚未运行；一旦出现故障，回溯数据库历史状态更是困难重重。
为何不直接使用 Alembic
谈到Python数据库迁移工具，Alembic无疑是生态中最成熟的选择。它与SQLAlchemy深度集成，功能全面。但实际使用中存在一定门槛：需要理解env.py、alembic.ini配置和版本链等概念；其autogenerate功能虽能自动生成迁移脚本，但往往需要人工仔细审查，结果并不总是符合预期；更重要的是，对于未采用SQLAlchemy ORM、直接使用原生SQL的项目而言，引入完整的Alembic体系可能显得过于笨重。
因此，本文借鉴Alembic最核心的两个设计理念——迁移文件版本化与执行历史持久化——构建了一套更轻量的解决方案。该方案基于纯Python和原生SQL，无需额外依赖，文件结构清晰直观，团队新成员几乎无需学习即可快速上手。下面详细介绍这套方案的设计思路与具体实现。
核心功能概述
用一句话概括：将每一次数据库结构变更，转化为具备版本记录、可追溯且不会重复执行的代码提交。
具体而言，它能帮助您实现：
- 每次表结构修改都编写为独立的迁移文件，提交至代码仓库，并与业务代码一同进行代码审查。
- 迁移执行器自动记录已执行的迁移，确保不会重复运行。
- 新成员拉取代码后，只需运行一条命令，即可将数据库自动同步至最新一致状态。
- 保持本地、测试、生产等多环境数据库状态一致，无需依赖人工记忆与手动对齐。
项目目录结构
migrations/
user.py # 用户表迁移脚本
orders.py # 订单表迁移脚本
coupons.py # 优惠券表迁移脚本
migrate.py # 迁移执行器（位于项目根目录）
结构非常扁平，无复杂嵌套。每张业务表对应一个迁移文件，核心执行器脚本直接置于项目根目录。
迁移文件格式详解
每个迁移文件主要定义两部分内容：需要执行的SQL语句列表，以及执行完成后的校验SQL。
sql = [
{
'id': 1, # 迁移编号，同一文件内唯一且只增不改
'sql': """
DROP TABLE IF EXISTS `orders`;
CREATE TABLE `orders` (
`id` BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY,
`order_no` VARCHAR(32) NOT NULL UNIQUE,
`user_id` INT NOT NULL,
`total_amount` DECIMAL(12,2) NOT NULL,
`order_status` TINYINT NOT NULL DEFAULT 0,
`create_time` DATETIME NOT NULL
);
""",
},
]
checks = [
'SELECT COUNT(*) FROM `orders`', # 验证表存在且可正常查询
]
# ── 执行器兼容格式，请勿修改 ──
migrations = [
{**item, 'checks': checks}
for item in sql
]
这里有几点关键设计值得注意：
首先是id字段，必须遵循只增不改原则。每条迁移上线后，其id即相当于唯一标识，执行器完全依赖它来判断迁移是否已执行。若修改已上线迁移的id，将导致执行器无法识别，可能引发重复执行风险。
其次是追加而非修改原则。当需要为orders表新增字段时，正确做法不是修改id: 1对应的SQL，而是在文件末尾追加一个id: 2的新条目：
sql = [
{ 'id': 1, 'sql': "..." }, # 已执行，保持不动
{
'id': 2,
'sql': """
ALTER TABLE `orders`
ADD COLUMN `source` TINYINT DEFAULT 0 COMMENT '订单来源';
""",
},
]
此规则确保了迁移历史的线性与可追加性，使得在任何时间点都能准确重现数据库的当时状态。
最后是checks字段，它作为最后一道安全屏障。迁移执行完毕后，执行器会自动运行checks列表中的SQL进行验证。通常只需编写简单的SELECT COUNT(*)即可，主要目的是确认表已成功创建且可正常查询，避免因语法错误等原因导致建表失败却未被察觉的情况。
使用方法指南
使用非常简单，主要通过执行器脚本的两个命令：
查看当前迁移状态——快速了解哪些迁移已执行、哪些待执行：
python3 migrate.py --status
执行所有待执行的迁移：
python3 migrate.py
执行器会按照文件顺序及每个文件内的id顺序依次执行迁移，已标记为“已执行”的会自动跳过，仅执行新增部分。
执行记录存储机制
本方案的核心在于状态管理。执行器会在您的数据库中自动创建一张名为_migration_history的表，专门用于记录每条迁移的执行状态：
| 字段 | 说明 |
|---|---|
| module | 迁移文件名（不含 .py 后缀），如 orders |
| migration_id | 迁移条目的 id |
| description | 迁移描述 |
| executed_at | 执行时间戳 |
此表即为“已执行迁移”的权威记录。下次运行迁移命令时，执行器会查询此表，所有module + migration_id组合已存在的记录均会被跳过，仅执行新的迁移。整个过程无需人工维护，状态完全由工具自动化管理。
示例项目中的三张核心表
为具体说明，本方案示例中管理了三张核心业务表：
users（用户表）：存储用户基本信息与积分，特别设计了added_by和updated_by审计字段，便于追踪数据的创建与修改者。
orders（订单主表）：覆盖订单从创建到完成的完整生命周期。order_status字段使用TINYINT枚举各种状态（如待支付、已完成、已取消）。同时，为支持“按用户+状态+时间”查询的高频场景，建立了组合索引idx_user_status_time。
coupons（优惠券表）：支持满减（fixed）与折扣（percent）两种优惠类型，通过外键关联users.id，并利用min_amount字段控制优惠券使用门槛。
这三张表的所有结构变更，均通过上述迁移文件进行管理。在任何环境中，只需执行python3 migrate.py，即可将数据库同步至一致的最新状态。
新增迁移的完整操作流程
- 在
migrations/目录下，找到对应的.py迁移文件（若不存在则新建）。 - 在该文件的
sql列表末尾，追加一个新条目，并确保id递增。 - 运行
python3 migrate.py执行迁移。
更稳妥的操作习惯是，先查看状态确认无误再执行：
# 先查看状态，确认预期迁移 python3 migrate.py --status # 确认无误后执行 python3 migrate.py
提交代码时，请将新增的迁移文件与业务代码一同提交。其他环境（如测试、生产）在拉取代码后，只需运行一遍迁移命令，数据库结构便会自动对齐，极大降低了沟通与操作成本。
与原生 Alembic 的主要区别
本方案虽借鉴了Alembic的核心理念，但在实现上更为直接与轻量，主要区别如下：
| 原生 Alembic | 本方案 | |
|---|---|---|
| 依赖 | SQLAlchemy + Alembic | 纯 Python，无额外依赖 |
| 迁移文件 | 自动生成，带版本哈希 | 手写 SQL，结构固定 |
| 历史记录 | alembic_version 表 |
_migration_history 表 |
| 版本管理 | 链式版本图 | 线性 id 追加 |
| 适用场景 | SQLAlchemy ORM 项目 | 任何使用原生 SQL 的项目 |
| 上手成本 | 需理解版本链概念 | 阅读本文后即可上手 |
若您的项目已深度使用SQLAlchemy ORM，直接采用Alembic无疑是更合适、更强大的选择。而本方案更适用于未使用ORM、直接编写原生SQL、且希望迁移流程尽可能简单可控的场景。
总结
数据库变更管理越早规范，后期维护越省心。等到线上出现“某个环境缺失字段”等问题时，排查往往耗时费力。
回顾而言，这套轻量方案的核心逻辑非常清晰，仅有三条：
- 变更写成文件，提交至仓库，让数据库结构与代码一样接受版本控制。
- id 只增不改，确保每一次变更历史都可追溯、可重现。
- 执行器自动记录状态，彻底摆脱依赖人脑记忆执行历史的不可靠方式。
其结构虽简单，却精准地堵住了数据库变更中最易出错的环节，以最小成本与复杂度，为项目提供了可靠的数据库迁移保障。
