📚 课程目录第4章数据库迁移

4. 数据库迁移

在批号跟踪接口开发中，ERP 与 MES 的数据模型会随着业务需求不断演进。手动修改数据库表结构不仅容易出错，更会导致开发、测试、生产环境之间的数据模型不一致。Alembic 作为 SQLAlchemy 的数据库迁移工具，能够以代码化、可追溯的方式管理数据库 Schema 变更，是实现持续交付的基础设施。

4.1 为什么需要数据库版本控制

没有版本控制的数据库变更通常表现为：开发人员在本地直接执行 ALTER TABLE，测试环境手动重复操作，生产环境由 DBA 编写脚本执行。这种模式存在三个致命缺陷：

变更不可追溯：无法确认某个字段是何时、由谁、基于哪个需求添加的。
环境差异累积：开发库、测试库、生产库的 Schema 逐渐偏离，导致“在我机器上能跑”的经典问题。
回滚成本极高：一旦上线后发现问题，只能手动逆向修改，极易造成数据丢失。

  核心原则：数据库 Schema 应当像应用代码一样，纳入 Git 版本管理，并通过自动化工具执行变更。Alembic 正是这一原则在 Python/SQLAlchemy 生态中的标准实现。

4.2 Alembic 核心概念与架构

Alembic 的迁移体系由三个核心组件构成，理解它们之间的关系是正确使用的前提。

组件	作用	对应文件/目录
迁移环境	定义数据库连接、元数据对象、迁移脚本加载路径	`alembic.ini` + `env.py`
迁移脚本	每个版本对应一个 Python 脚本，包含 `upgrade()` 和 `downgrade()` 函数	`versions/xxxx_xxx.py`
版本表	记录当前数据库所处的迁移版本号，确保迁移幂等执行	`alembic_version`（数据库表）

下图展示了 Alembic 在数据库迁移过程中的工作流程与组件协作关系：

4.3 初始化 Alembic 环境

在项目根目录执行以下命令完成初始化：

# 安装 Alembic
pip install alembic

# 初始化迁移环境
alembic init alembic

初始化后生成的项目结构如下：

your_project/
├── alembic/
│   ├── env.py              # 迁移环境配置，定义 SQLAlchemy 元数据
│   ├── script.py.mako      # 迁移脚本模板
│   └── versions/           # 存放所有迁移脚本
├── alembic.ini             # 数据库连接字符串等全局配置
└── models/                 # 你的 SQLAlchemy 模型定义

关键配置：在 alembic/env.py 中，必须将 target_metadata 指向你的 SQLAlchemy Base 元数据，否则 --autogenerate 无法检测模型变更。

4.4 编写初始迁移脚本

假设批号跟踪系统需要创建 batch_trace 和 batch_operation 两张核心表，对应的 SQLAlchemy 模型如下：

# models/batch_models.py
from sqlalchemy import Column, Integer, String, DateTime, ForeignKey, Text
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

Base = declarative_base()

class BatchTrace(Base):
    __tablename__ = 'batch_trace'
    id = Column(Integer, primary_key=True, autoincrement=True)
    batch_no = Column(String(50), unique=True, nullable=False, comment='批号')
    material_code = Column(String(50), nullable=False, comment='物料编码')
    quantity = Column(Integer, nullable=False, comment='数量')
    created_at = Column(DateTime, default=datetime.utcnow, comment='创建时间')

class BatchOperation(Base):
    __tablename__ = 'batch_operation'
    id = Column(Integer, primary_key=True, autoincrement=True)
    batch_id = Column(Integer, ForeignKey('batch_trace.id'), nullable=False, comment='关联批号ID')
    operation_code = Column(String(30), nullable=False, comment='工序编码')
    operator = Column(String(50), comment='操作人')
    operate_time = Column(DateTime, default=datetime.utcnow, comment='操作时间')
    remark = Column(Text, comment='备注')

生成初始迁移脚本：

alembic revision --autogenerate -m "create_batch_trace_tables"

该命令会自动对比当前数据库 Schema 与 SQLAlchemy 模型的差异，在 versions/ 目录下生成类似 xxxx_create_batch_trace_tables.py 的脚本。生成的 upgrade() 函数内容大致如下：

def upgrade():
    op.create_table('batch_trace',
        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
        sa.Column('batch_no', sa.String(length=50), nullable=False),
        sa.Column('material_code', sa.String(length=50), nullable=False),
        sa.Column('quantity', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('batch_no')
    )
    op.create_table('batch_operation',
        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
        sa.Column('batch_id', sa.Integer(), nullable=False),
        sa.Column('operation_code', sa.String(length=30), nullable=False),
        sa.Column('operator', sa.String(length=50), nullable=True),
        sa.Column('operate_time', sa.DateTime(), nullable=True),
        sa.Column('remark', sa.Text(), nullable=True),
        sa.ForeignKeyConstraint(['batch_id'], ['batch_trace.id'], ),
        sa.PrimaryKeyConstraint('id')
    )

4.5 执行与回滚迁移

迁移脚本编写完成后，通过以下命令控制数据库版本：

操作	命令	说明
查看当前状态	`alembic current`	显示数据库当前所在的迁移版本
查看历史	`alembic history`	列出所有迁移版本及其依赖关系
升级到最新	`alembic upgrade head`	执行所有尚未应用的迁移脚本
升级指定版本	`alembic upgrade <revision_id>`	迁移到指定的版本号
回滚一个版本	`alembic downgrade -1`	回退到上一个迁移版本
回滚到指定版本	`alembic downgrade <revision_id>`	回退到指定的历史版本

执行初始迁移：

# 将数据库升级到最新版本（即执行初始迁移）
alembic upgrade head

执行成功后，数据库中会创建 batch_trace、batch_operation 两张业务表，以及 alembic_version 版本控制表。验证方法：

# 查看当前版本
alembic current

# 预期输出：xxxx_create_batch_trace_tables (head)

如果发现初始模型设计有误，需要回滚：

# 回滚到上一个版本（即空库状态）
alembic downgrade -1

回滚风险提示：downgrade() 函数通常执行 DROP TABLE 操作，会永久删除表和数据。在生产环境中执行回滚前，务必确认 downgrade() 的实现逻辑，并提前做好数据备份。建议在开发/测试环境充分验证后再操作生产库。

4.6 迁移最佳实践

在 ERP 与 MES 批号跟踪接口的长期维护中，遵循以下规范可以显著降低数据库变更风险：

每次模型变更对应一个迁移脚本：不要在一个脚本中混合多个不相关的 Schema 变更，保持迁移脚本的原子性。
迁移脚本必须经过 Code Review：自动生成的脚本可能存在误判（如字段类型推断错误），人工审查 upgrade() 和 downgrade() 逻辑是必要环节。
禁止手动修改 alembic_version 表：该表由 Alembic 自动维护，手动插入或删除版本记录会导致迁移状态混乱。
迁移脚本纳入 Git 版本管理：所有 versions/ 目录下的 .py 文件必须提交到代码仓库，确保团队成员共享同一套迁移历史。
CI/CD 中自动执行迁移：在持续集成流水线中加入 alembic upgrade head 步骤，确保测试环境始终与模型定义同步。

下图总结了数据库迁移在批号跟踪接口开发中的完整生命周期：

通过以上步骤，批号跟踪接口的数据库变更实现了完整的版本控制闭环。无论是新增追溯字段、调整工序表结构，还是添加索引优化查询性能，每一次变更都有迹可循、可回滚、可审计，为 ERP 与 MES 的数据一致性提供了坚实基础。