Python中实现数据库迁移的轻量级方案详解_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	迁移文件名（不含 .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 的项目
上手成本	需要理解版本链概念	看完本文即可上手