Flyway 免费版：用 Python 实现回滚功能

# aconcan 来源：https://aconcan.io/blog/flyway-for-freeloaders/ *一条命令回滚已应用的迁移，无需Flyway商业版授权，无需`undo`命令，无需在俱乐部哭泣\.* 如果你曾因为`undo`命令而考虑购买Flyway团队版，你一定懂我的意思。托管式回滚功能是Flyway付费墙后最诱人的那块蛋糕，也是每个人在凌晨两点遇到生产环境迁移崩溃时最渴望的功能。社区版只提供`migrate`、`info`、`validate`、`repair`，当你要求它执行回滚时，它只会把烟头摁灭在你眼睛里。 解决方案？其实你根本不需要付费版的`undo`，只需要对“回滚”的真正*含义*稍作巧妙的变通。以下回滚方案完全通过Flyway命令实现，无需通过独立驱动程序直接执行任何SQL（#极简主义#）。下面是具体原理，坐稳了各位。坐稳了 ## 核心概念：回滚只是穿了伪装的迁移 关键突破点在于：Flyway社区版会愉快地执行你提供的任何版本化迁移。它不关心这个迁移是创建表还是删除表，SQL就是SQL。不过，Flyway对两件事非常执着，而我们的方案正是基于它们：**版本号必须始终递增**和**schema历史记录表是铁律**。 因此，我们不是要求Flyway反向执行`V2`，而是编写一个*新的*、更高版本号的迁移，其内容恰好是`V2`的逆操作，并让Flyway*向前*迁移到这个版本。基本上就是把“撤销”操作表达为`migrate`命令，数据库最终会恢复到初始状态。然后我们在`flyway_schema_history`表中清理迁移记录及其对应的“撤销”迁移，使其与数据库状态和磁盘文件重新对齐。 ## 磁盘文件结构 每个作用域需要两个目录：预定的迁移目录，以及包含反向逻辑的并行“down”脚本目录。结构如下： ``` migrations/ V1__create_customers.sql rollbacks/ V1__create_customers.down.sql ``` 命名规范至关重要。“down”脚本应通过`V__`前缀与其对应的初始迁移匹配。在回滚过程中，down脚本的版本号会递增，因此Flyway会将其视为新的待应用迁移（稍后详述）。我建议提前验证每个迁移是否有对应的“down”脚本，避免回滚中途出现缺失： ```python def _down_file(rollbacks: Path, version: str) -> Path: matches = sorted(rollbacks.glob(f"V{version}__*.down.sql")) if not matches: raise RollbackError(f"no down script for V{version} in {rollbacks}") return matches[0] ``` ## 我们到底在回滚什么？ 一个重要问题：当有人执行`rollback`时，他们具体要回滚什么？最近一次迁移？最近一批迁移？如果最后一次`migrate`同时应用了三个文件怎么办？ 解决方案是：每次执行`flyway migrate`时，记录成功应用的迁移批次。这通过比较`migrate`前后已应用的版本差异实现，并将新应用的版本存储到JSON文件中。基本上，所有新版本就构成“该批次”。 ```python def migrate(config: PgConfig, scope: str) -> int: # 使用 info -outputType=json 获取当前已应用版本 before = set(flyway.applied_versions(config)) # 执行迁移 code = flyway.migrate(config, [config.migrations]) if code != 0: return code # 获取迁移前后已应用版本的差异 after = flyway.applied_versions(config) batch = [v for v in after if v not in before] # 存储到JSON状态文件 if batch: state.write_batch(scope, batch) return 0 ``` 如果你通过进程驱动的CLI方式执行此操作，请将状态文件写入容器内能在独立的`migrate`和`rollback`调用之间保留的路径。`if batch:`中隐藏着一种防御性逻辑：如果意外运行两次`migrate`，第二次不会应用新内容，`batch`为空，记录批次不会被覆盖为空列表`[]`。 ## 准备回滚脚本 既然我们已持久化记录上次`migrate`调用的应用内容，就确切知道要回滚什么。现在可以开始准备带有修改版本号的“down”迁移，以满足Flyway“版本号必须递增”的条件。 给定批次`[1, 2]`，我们构建一个临时目录，包含仅供Flyway使用的合成迁移： ```python def _stage(config: PgConfig, batch: list[str], staging: Path) -> None: # 获取已应用的最高版本号 highest = max(int(v) for v in flyway.applied_versions(config)) # 反转批次顺序，并从当前最高版本号开始递增编号 undo_versions = [] for offset, version in enumerate(reversed(batch), start=1): new_version = str(highest + offset) undo_versions.append(new_version) sql = _down_file(config, version).read_text() (staging / f"V{new_version}__undo_V{version}.sql").write_text(sql) ``` 现在我们以Flyway易于应用的格式准备好了所有用于回滚的“down”文件。有两点值得注意： **1. 我们反转批次顺序。**如果先应用`V1`再应用`V2`，则必须先撤销`V2`才能撤销`V1`。因此最近应用的迁移最先被撤销，并获得*最低*的新版本号。批次`[1, 2]`当前最高水位为`2`时变为： ``` V3__undo_V2.sql V4__undo_V1.sql ``` **2. 版本号持续递增。**我们从不重复使用`1`或`2`。Flyway的基本原则是版本号只能递增，因此撤销脚本的编号*高于*当前最大版本号。这让我们能够以普通正向迁移的形式，绕过社区版的限制执行逆向操作。 ## 对齐`flyway_schema_history`表 如果仅止步于此，在准备“down”脚本并运行`migrate`后，`flyway info`会显示`V3__undo_V2`和`V4__undo_V1`的记录，尽管这些迁移在我们的迁移目录中并不存在（且尽管它们抵消了数据库中先前迁移的逻辑，使其恢复初始状态）。 由于这些迁移没有对应文件，Flyway会标记历史记录不同步，并在后续的`validate`和`migrate`命令中报错。因此，我们最后在临时目录中放入一个`afterMigrate.sql`回调脚本，用于从历史表中删除往返记录： ```sql -- 删除原始迁移和对应"down"迁移的记录 DELETE FROM {config.schema}.flyway_schema_history WHERE version IN ({versions}); ``` 我们删除往返操作的*双方*记录：原始版本（`1`、`2`）**和**撤销条目（`3`、`4`）。尘埃落定后，历史表会完全忘记这一切。`customers`表消失，`V1__create_customers.sql`仍在磁盘上，`flyway info`会愉快地将其报告为`Pending`（待处理），准备重新应用，就像你从未执行过迁移一样。 `afterMigrate`在Flyway自身的执行流程中运行，在版本化迁移成功后触发，且仅在迁移成功时执行。 ## 实现原子性 你想要回滚功能的根本原因就是事情可能出错。因此回滚操作本身*绝不能*让你处于半完成状态。这里我们依赖Flyway的一个标志： ```python def migrate(config: PgConfig, locations: list[str]) -> int: # 作为单个数据库事务运行。如果任何待处理迁移失败，整个批次回滚。 args = _base_args(config, locations) + ["-group=true", "migrate"] return subprocess.run(args).returncode ``` `-group=true`将所有待处理迁移包装在单个事务中。要么所有“down”脚本和清理回调一起提交，要么PostgreSQL回滚整个操作，数据库不受影响。不会出现`V2`已撤销但`V1`未撤销的中间状态。（这要求你的DDL操作是事务性的） 组装好的`rollback`函数将一切整合： ```python def rollback(config: PgConfig, scope: str) -> int: # 从状态文件读取刚应用的迁移版本批次 batch = state.read_batch(scope) if not batch: raise RollbackError(f"no recorded migration batch to undo for {scope}") # 验证每个down脚本都存在 for version in batch: _down_file(config, version) # 准备临时版本化的down脚本和回调，然后调用migrate with tempfile.TemporaryDirectory() as tmp: _stage(config, batch, Path(tmp)) code = flyway.migrate(config, [config.migrations, tmp]) # 成功后清除状态文件中刚应用的迁移记录（初始批次已回滚） if code == 0: state.clear_batch(scope) return code ``` 这里的执行顺序是刻意设计的： 1. **无批次则硬失败** 没有记录却执行回滚是错误的。如果回滚不在当前会话中进行，很可能是在回滚已存在一段时间的先前迁移；这种情况下应该向前修复而非回滚。 2. **在操作数据库*前*验证每个“down”脚本存在** 如果批次中任何一个版本缺少`.down.sql`文件，就必须失败终止。就像膝盖中箭一样，半途而废的回滚比完全不回滚更糟糕。 3. **跨越两个位置执行migrate** 注意Flyway需要真实的迁移目录***和*临时准备目录**。Flyway需要原始文件在位置中协调历史记录，而临时目录提供新的down版本。 4. **仅在成功时清除状态** 如果migrate返回非零状态，批次文件保持不变，以便你修复问题并重试。临时目录随`with`块结束自动销毁。完成后临时目录及其所有内容自动删除。Flyway版本的“down”文件仅在单次Flyway调用期间存在，之后立即消失。 ## 局限性 说实话，我认为这个方案没有太多复杂部分，但值得提及一些限制，以了解其边界： - **方案未处理数据丢失：**这会让你处于*与实际购买Flyway团队版相同的位置*，但我必须提及，因为透明是我的原则。如果你的“up”脚本执行了不可逆操作（如删除包含数据的列），希望你真的不需要那些数据。 - **“down”脚本需要手动编写：**没有自动生成的逆操作，但即使购买Flyway团队版*你仍然需要手动编写*。 - **状态保存在容器文件系统上：**回滚上下文与文件系统生命周期绑定。在migrate和rollback之间销毁容器会丢失批次记录。（这不适用于无状态环境） - **我们从`flyway_schema_history`中删除了记录：**Flyway纯粹主义者会脊背发凉。我对这感到满意，因为删除范围精确限定于往返操作中的版本，且在同一事务中运行。但请注意，你正在Flyway不知情的情况下修改其簿记操作。要明确知道你在做什么。 ## 为何这实际上很好 尽管这个方案像是两个Flyway操作在套娃，但结果是诚实的回滚。它通过相同的引擎执行逆向SQL，具有与正向迁移相同的事务保证。修改`flyway_schema_history`表通过Flyway支持的功能完成。只要你使用数据库事务，就能很好地防止各种半成品操作。 完成后，你的schema历史记录保持干净，被回滚的迁移会重新变为待处理状态，准备重新应用（或我假设是进行修复）。无需授权。无需`undo`命令。只需一个临时目录、一个只递增的版本计数器，以及一个漂亮的`DELETE`语句。 一个流传千年的故事。