Claude 工作流实验室
独立第三方实践站 · 非 Anthropic 官方;域名中的 6 不代表官方模型版本独立中文指南
验证与复盘

用 Claude 设计可回滚的 SQLite 迁移:先演练失败路径

用 Python 标准库为文章表新增状态字段,验证失败后结构和版本都回滚,再测试重复执行行为。

数据库迁移最危险的时刻,往往不是语法错误,而是“前半段成功,后半段失败”。让 Claude 生成 SQL 时,除了目标表结构,还应交代当前版本、原数据要怎样保留,以及失败后必须恢复到什么状态。下面只在内存数据库演练,不触碰真实业务文件。

一 把迁移契约说完整

原表为 notes(id, title)。本次新增 status,只允许 draft 或 published,旧记录统一为 draft。版本从 0 升到 1;遇到未知版本立即停止;已经是 1 时不重复加列。一次迁移中的表结构变更与版本更新必须共同成功或共同撤销。

真实环境还应先备份,并验证备份能恢复。复制正在写入的数据库文件可能遗漏相关状态,应根据应用情况使用 SQLite 的备份能力。Python 官方 sqlite3 文档介绍了事务控制与连接备份接口。本文代码要求 Python 3.12 或以上,以便显式使用 autocommit 参数。

二 完整的失败演练

保存为 migration_demo.py,运行 python3 migration_demo.py:

python 示例代码
import sqlite3


def migrate(db, fail=False):
    db.execute('BEGIN IMMEDIATE')
    try:
        version = db.execute('PRAGMA user_version').fetchone()[0]
        if version == 1:
            db.execute('COMMIT')
            return False
        if version != 0:
            raise RuntimeError('不支持的数据库版本')
        db.execute("ALTER TABLE notes ADD COLUMN status TEXT "
                   "NOT NULL DEFAULT 'draft' "
                   "CHECK(status IN ('draft', 'published'))")
        if fail:
            raise RuntimeError('模拟迁移中断')
        db.execute('PRAGMA user_version = 1')
        db.execute('COMMIT')
        return True
    except Exception:
        if db.in_transaction:
            db.execute('ROLLBACK')
        raise


db = sqlite3.connect(':memory:', autocommit=True)
db.execute('CREATE TABLE notes(id INTEGER PRIMARY KEY, title TEXT NOT NULL)')
db.execute('INSERT INTO notes(title) VALUES (?)', ('测试文章',))
try:
    migrate(db, fail=True)
except RuntimeError:
    pass
assert [r[1] for r in db.execute('PRAGMA table_info(notes)')] == ['id', 'title']
assert db.execute('PRAGMA user_version').fetchone()[0] == 0
assert migrate(db) is True
assert db.execute('SELECT title, status FROM notes').fetchall() == [('测试文章', 'draft')]
assert db.execute('PRAGMA user_version').fetchone()[0] == 1
assert migrate(db) is False
try:
    db.execute("UPDATE notes SET status = 'invalid'")
    raise AssertionError('非法状态没有被拒绝')
except sqlite3.IntegrityError:
    pass
db.close()
print('migration and rollback checks passed')

示例显式发送 BEGIN、COMMIT 与 ROLLBACK,避免把事务边界藏进默认行为。失败发生在加列之后、版本更新之前,因此断言能够验证结构与版本一起回到原状。没有使用会让事务边界更难读的多语句脚本接口。

三 给 Claude 的迁移提示词

可复制提示词
请为 SQLite 设计从版本 0 到版本 1 的迁移,使用 Python 3.12 标准库。
原表:notes(id INTEGER PRIMARY KEY, title TEXT NOT NULL)。
新增 status,仅允许 draft/published,旧记录为 draft;版本用 PRAGMA user_version。
要求结构和版本原子更新;未知版本停止;重复执行版本 1 时无操作。
先给迁移前检查、备份与恢复演练清单,再给内存数据库可运行例子。
必须模拟加列后失败,并断言结构、数据和版本正确回滚。
不连接生产数据库,不生成自动删除备份或覆盖原文件的命令。

如果已有外键、触发器、索引或视图,必须连同定义一起提供。新增一列的示例不能直接推广为删列、改主键或重建大表;这些操作的锁定时间和兼容性需要额外设计。

四 上线前检查什么

本文在 Python 3.12.14 的内存数据库中通过了失败回滚、默认值、版本更新、重复执行与约束拒绝检查;没有对生产数据库或磁盘故障做测试。实际迁移前,应确认应用暂时停止写入或已有适合的并发策略,记录 SQLite 运行时版本,并在备份副本上检查行数与关键数据。

事务回滚也不等于应用版本回退。迁移已经提交后,旧程序是否还能读新结构,需要单独验证;若用户已产生新数据,直接恢复旧备份会丢失这些写入。把数据库回滚、应用回退和备份恢复分开写清,才是一份可执行的迁移方案。

五 锁等待与版本检查不能省略

BEGIN IMMEDIATE 获取写事务时可能遇到其他写入者,失败并不必然说明迁移 SQL 有错。应先判断锁竞争来源,再按维护窗口与应用策略处理,避免无限重试。重试前重新检查数据库版本;若另一个进程已经完成迁移,盲目重复加列会产生新的故障。

本例把 user_version 留给这一个迁移流程。如果项目已有迁移框架,先确认它怎样记录版本,再让 Claude 生成符合现有机制的脚本,不要另占同一个版本字段。正式验收还应查询表结构和约束是否与预期相符,不能只看版本数字已经变成一。版本值是一条记录,不是数据库结构正确的独立证明。

方法参考:Anthropic 提示工程概览。