Python中实现数据库迁移的轻量级方案详解

项目上线后，数据库的改动往往是风险最高的环节之一。加个字段、改个索引、新建一张表——听起来都是小事，但实际操作中，团队或多或少都踩过类似的坑：本地改了，测试环境忘了同步；测试环境执行了，生产环境又漏了一条ALTER语句；多人协作时，根本记不清哪条SQL执行过、哪条还没跑；一旦出了问题想回溯，完全不知道数据库当时是什么状态。

为什么不直接用 Alembic

提到Python的数据库迁移，Alembic无疑是生态中最成熟的选择，它与SQLAlchemy深度集成，功能非常完整。但用过的人都知道，它有一定的上手门槛：你需要理解env.py、alembic.ini、版本链等一系列概念；其autogenerate功能虽然方便，但自动生成的迁移文件往往需要人工仔细审查，结果并不总是符合预期；更重要的是，对于那些没有使用SQLAlchemy ORM、而是直接编写原生SQL的项目来说，引入Alembic整套体系反而显得有些笨重和过度设计。

因此，这里借鉴了Alembic最核心的两个设计思想——迁移文件版本化和执行历史持久化——自己搭建了一套更轻量的方案。它基于纯Python和原生SQL，没有任何额外依赖，文件结构一目了然，团队新人几乎不需要学习成本就能立刻上手。下面就来详细介绍这套方案的设计思路和具体用法。

它能做什么

用一句话概括就是：将数据库的每一次结构变更，都转化为有版本记录、可追溯、且不会重复执行的代码提交。

具体来说，它能帮你实现：

• 每次修改表结构，都写成独立的迁移文件，提交到代码仓库，并和业务代码一起进行Code Review。
• 迁移执行器会自动记录哪些迁移已经执行过，确保下次不会重复运行。
• 新同事拉取代码后，只需运行一条命令，数据库就能自动同步到最新的一致状态。
• 本地、测试、生产等多环境之间的数据库状态保持一致，不再依赖人工记忆和手动对齐。

目录结构

migrations/

user.py # users 表迁移

orders.py # orders 表迁移

coupons.py # coupons 表迁移

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 + 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，就能将数据库同步到一致的最新状态。

新增迁移的完整流程

1. 在migrations/目录下，找到对应的.py迁移文件（如果不存在则新建一个）。
2. 在该文件的sql列表末尾，追加一个新的条目，并确保id是递增的。
3. 运行python3 migrate.py执行迁移。

一个更稳妥的操作习惯是，先查看状态确认无误再执行：

# 先看一眼状态，确认预期
python3 migrate.py --status
# 没问题就执行
python3 migrate.py

提交代码时，记得将新增的迁移文件和业务代码一同提交。其他环境（如测试、生产）在拉取代码后，只需要运行一遍迁移命令，数据库结构就会自动对齐，极大减少了沟通和操作成本。

和原生 Alembic 的区别

这套方案虽然借鉴了Alembic的核心理念，但在实现上更加直接和轻量，主要区别如下：

如果你的项目已经深度使用了SQLAlchemy ORM，那么直接采用Alembic无疑是更合适、更强大的选择。而这套方案则更适用于那些没有使用ORM、直接编写原生SQL、并且希望迁移流程尽可能简单可控的场景。

小结

数据库变更管理这件事，越早规范，后期就越省心。等到线上真的出现“某个环境少了个字段”这类问题，排查起来往往耗时费力。

回顾一下，这套轻量方案的核心逻辑其实非常清晰，只有三条：

1. 变更写成文件，提交到仓库，让数据库结构和代码一样接受版本管理。
2. id 只增不改，确保每一次变更历史都可追溯、可重现。
3. 执行器自动记录状态，彻底摆脱依赖人脑记忆执行历史的不可靠方式。

它的结构虽然简单，却精准地堵住了数据库变更中最容易出错的几个环节，用最小的成本和复杂度，为项目提供了可靠的数据库迁移保障。