IoT 后端的 SQLite 增量迁移——没上 Alembic，纯手工干了 20 多个补丁

这个项目用的 Peewee + SQLite。Peewee 自带 migrate() 模块，也有 Alembic 这样的第三方方案。但我们一概没用——不是不知道，是试了之后发现，IoT 后端的迁移场景跟 Web 应用不一样：字段多、关联杂、中间表是 Peewee 自动生成的，而且经常要在已有数据上修索引、回填默认值。

Alembic 每次生成一个迁移脚本文件，时间戳命名，堆了几十个文件之后，部署到现场设备上——树莓派的 SD 卡读写本来就慢，迁移脚本跑一半断电了，数据库就废了。

最后选了最朴素的方案：启动时跑一遍 apply_migrations()，纯 SQL，幂等，不加版本号。跑了半年，20 多个补丁，一次没翻车。

核心思路：先检查再执行，每步都幂等

整个迁移函数的结构就三层：

def apply_migrations(db):
    # 第一层：加字段
    changed |= add_column_if_missing("entity", "extra", "TEXT")
    changed |= add_column_if_missing("rover", "runtime_config", "TEXT")
    changed |= add_column_if_missing("rover", "config_version", "INTEGER DEFAULT 1")

    # 第二层：修索引
    changed |= fix_modbus_register_through_unique_index()
    changed |= ensure_runtime_config_version_unique_index()

    # 第三层：建新表
    if not table_exists("rover_runtime_config_version"):
        RoverRuntimeConfigVersion.create_table(safe=True)

    # 第四层：回填数据
    db.execute_sql("UPDATE rover SET config_version = 1 WHERE config_version IS NULL OR config_version <= 0")

每一层都自己检查”要不要做”，不依赖外部状态。add_column_if_missing 检查列是否存在，table_exists 检查表是否存在，索引修复检查索引结构是否正确。

加字段：SQLite 的 ALTER TABLE 只有 ADD COLUMN

SQLite 的 ALTER TABLE 支持非常有限：只能 ADD COLUMN，不能 DROP COLUMN（3.35.0 之前），不能 MODIFY COLUMN，不能加 UNIQUE 约束。

所以加字段是最安全的操作：

def add_column_if_missing(table_name, column_name, column_def):
    if column_exists(table_name, column_name):
        return False
    db.execute_sql(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_def}")
    return True

column_exists 用的是 PRAGMA table_info：

def column_exists(table_name, column_name):
    rows = db.execute_sql(f"PRAGMA table_info('{table_name}')").fetchall()
    return any(row[1] == column_name for row in rows)

注意 column_def 里要带默认值，比如 INTEGER DEFAULT 1。SQLite 的 ADD COLUMN 对新行生效，已有行的该列值是 NULL——除非你显式指定了 DEFAULT。但即使指定了 DEFAULT，已有行仍然是 NULL（SQLite 的怪癖），所以后面还得跟一条 UPDATE 回填。

修索引：那个 Peewee 自动生成的中间表

这是最头疼的一个。Peewee 的 ManyToManyField 会自动创建中间表，表名是 modbus_device_config_modbus_device_register_through（对，就是这么长）。

早期代码没管唯一约束，Peewee 默认给中间表加了个单列唯一索引——只限制了 modbusdeviceconfig_id。这意味着同一个配置只能关联一个寄存器，但实际上一台 Modbus 设备有十几个寄存器。Insert 不会报错（因为索引只管唯一性不管存在性），但数据会悄悄丢。

修复分两步：先删错误索引，再建正确的联合索引。

def fix_modbus_register_through_unique_index():
    table_name = "modbus_device_config_modbus_device_register_through"
    if not table_exists(table_name):
        return False

    # 删除错误的单列唯一索引
    indexes = db.execute_sql(f"PRAGMA index_list('{table_name}')").fetchall()
    for idx in indexes:
        index_name = idx[1]
        is_unique = bool(idx[2])
        if not is_unique:
            continue
        cols = get_index_columns(index_name)
        if len(cols) == 1 and cols[0] == "modbusdeviceconfig_id":
            db.execute_sql(f"DROP INDEX IF EXISTS {index_name}")

    # 建联合唯一索引
    indexes = db.execute_sql(f"PRAGMA index_list('{table_name}')").fetchall()
    has_pair_unique = False
    for idx in indexes:
        cols = set(get_index_columns(idx[1]))
        if cols == {"modbusdeviceconfig_id", "modbusdeviceregister_id"}:
            has_pair_unique = True
            break

    if not has_pair_unique:
        db.execute_sql(
            f"CREATE UNIQUE INDEX IF NOT EXISTS idx_modbus_config_register_unique "
            f"ON {table_name}(modbusdeviceconfig_id, modbusdeviceregister_id)"
        )

为什么要先检查再建？因为迁移可能跑多次（重启就跑一次），不能假设索引一定不存在。CREATE UNIQUE INDEX IF NOT EXISTS 本身是幂等的，但前面的 DROP INDEX 不是——如果跑第二次，索引已经被删了，不需要再删。

表重命名：vl_recognition_task → inference_task

产品早期只有视觉识别任务，表叫 vl_recognition_task。后来加了其他推理任务，表名得改。SQLite 支持 ALTER TABLE ... RENAME TO：

if table_exists("vl_recognition_task") and not table_exists("inference_task"):
    db.execute_sql("ALTER TABLE vl_recognition_task RENAME TO inference_task")
    add_column_if_missing("inference_task", "task_type", "VARCHAR(255) DEFAULT 'vl_recognition'")
    add_column_if_missing("inference_task", "source_type", "VARCHAR(255)")
    add_column_if_missing("inference_task", "source_id", "INTEGER")
    add_column_if_missing("inference_task", "input_params", "TEXT DEFAULT '{}'")
    add_column_if_missing("inference_task", "result_data", "TEXT")
    db.execute_sql("UPDATE inference_task SET task_type = 'vl_recognition' WHERE task_type IS NULL OR task_type = ''")

关键条件：table_exists("vl_recognition_task") and not table_exists("inference_task")。两个表不能同时存在——如果 inference_task 已经存在，说明迁移已经做过了，跳过。如果两个都不存在，说明是新部署，建表由 db.create_tables() 处理，也跳过。

重命名之后加新字段、回填 task_type，一条龙搞定。老数据无缝过渡。

版本唯一索引修复：先清数据再建索引

rover_runtime_config_version 表需要 (rover_id, version_no) 的联合唯一索引。但历史数据里可能有重复版本号——早期代码没做校验，同一个 rover 的 version_no=1 出现了三行。

建唯一索引之前必须先去重，否则 CREATE UNIQUE INDEX 会失败：

def ensure_runtime_config_version_unique_index():
    # 先标准化异常版本号
    db.execute_sql(f"UPDATE {table_name} SET version_no = 1 WHERE version_no IS NULL OR version_no <= 0")

    # 查重
    duplicate_groups = db.execute_sql(
        f"SELECT COUNT(1) FROM ("
        f"  SELECT rover_id, version_no, COUNT(1) AS c"
        f"  FROM {table_name} GROUP BY rover_id, version_no HAVING c > 1"
        f")"
    ).fetchone()

    if dup_count > 0:
        # 同一 rover + version_no 只保留 id 最大的那一行
        db.execute_sql(
            f"DELETE FROM {table_name} WHERE id IN ("
            f"  SELECT t.id FROM {table_name} t"
            f"  JOIN ("
            f"    SELECT rover_id, version_no, MAX(id) AS keep_id"
            f"    FROM {table_name} GROUP BY rover_id, version_no HAVING COUNT(1) > 1"
            f"  ) d ON t.rover_id = d.rover_id AND t.version_no = d.version_no"
            f"  WHERE t.id <> d.keep_id"
            f")"
        )

保留 MAX(id) 那一行——id 自增，最大的就是最新的。删完重复数据，再建索引就稳了。

然后还要把 rover 表里的 config_version 同步成实际最大版本号：

db.execute_sql(
    "UPDATE rover SET config_version = COALESCE("
    "  (SELECT MAX(v.version_no) FROM rover_runtime_config_version v WHERE v.rover_id = rover.id),"
    "  CASE WHEN config_version IS NULL OR config_version <= 0 THEN 1 ELSE config_version END"
    ")"
)

COALESCE 兜底：如果版本历史表里有数据就取最大值，没有就保持原值，原值为空就默认 1。

回填默认值的那些 SQL

迁移的最后一步往往是回填。历史数据里新加的字段全是 NULL，业务代码如果没做空值判断就会崩。

db.execute_sql("UPDATE rover SET config_version = 1 WHERE config_version IS NULL OR config_version <= 0")
db.execute_sql("UPDATE rover SET rover_auth_enabled = 0 WHERE rover_auth_enabled IS NULL")
db.execute_sql(
    "UPDATE rover_model_artifact SET conversion_status='completed' "
    "WHERE conversion_status IS NULL OR conversion_status = ''"
)

这些 UPDATE 语句跑在 SQLite 上，几万行数据也就几百毫秒。如果是 MySQL/PostgreSQL 上百万行的大表，得考虑分批 UPDATE 避免锁表——但 IoT 后端部署在树莓派上，数据量不会超过几千行，一条 UPDATE 全搞定。

为什么不用 Alembic

试过，但遇到几个问题：

1. 现场设备断电：树莓派拔电源是常态。Alembic 迁移脚本执行到一半断电，alembic_version 表已经更新但 SQL 没跑完，重启后 Alembic 认为迁移已完成，实际上缺字段
2. Peewee 中间表：Alembic 不认识 Peewee 自动生成的中间表名，生成迁移脚本时会报错
3. 回填逻辑：Alembic 的 upgrade() 里写数据回填总觉得不太对劲，它设计上更适合改 schema 而不是改数据

我们的方案没版本号，每次启动全量检查。好处是天然幂等：断电了？重启再跑一遍，已经做过的操作会跳过。坏处是每次启动多几十毫秒的检查时间——在 IoT 场景里完全可以接受。

一点经验

• SQLite 的 PRAGMA table_info 和 PRAGMA index_list 是迁移的好帮手，比 try/except 优雅
• ALTER TABLE ADD COLUMN 带上 DEFAULT，减少回填 SQL 的数量
• 联合唯一索引建之前一定先查重复数据，否则 CREATE 会失败
• 表重命名用 ALTER TABLE RENAME TO，别新建表再 INSERT INTO SELECT——后者数据量大时锁表时间太长
• 每条迁移操作都假设自己可能被执行多次，写之前先检查

说白了，SQLite 迁移的核心原则就一个：别假设上一次迁移一定成功了。每一步都自己验证，该补就补，该跳就跳。