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

先说问题

项目上线之后，数据库改动是最容易出事的环节。

加一个字段、改一个索引、新建一张表——听起来很小的事，实际操作中经常踩坑：

• 本地改了，测试环境忘了改
• 测试改了，生产环境漏了一条 ALTER
• 多人协作，不知道哪条 SQL 执行过、哪条没执行
• 出了问题想回溯，根本不知道数据库当时是什么状态

为什么不直接用 Alembic

Alembic 是 Python 生态里最成熟的数据库迁移工具，和 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 = [
    {
        '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. 执行器自动记录状态，不靠人记忆哪条跑过

结构简单，但把最容易出错的环节都堵住了。

到此这篇关于Python中实现数据库迁移的轻量级方案详解的文章就介绍到这了,更多相关Python数据库迁移内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论