数据库迁移最危险的时刻,往往不是语法错误,而是“前半段成功,后半段失败”。让 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:
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 提示工程概览。