Flywayの無料版でロールバックをPythonで実装する

Flyway Community Editionで適用済みのマイグレーションをロールバックする単一コマンド、ライセンス不要、undoコマンド不要、クラブで泣く必要なし。 Flyway Teamsのundoコマンドのためだけに価格を調べたことがあるなら、私が何を言っているかわかるでしょう。管理されたロールバック機能は、Flywayの有料領域にある最も魅力的な機能であり、本番環境でマイグレーションが問題を起こした際に、深夜2時に誰もが欲しがる機能です。Community Editionは、migrate、info、validate、repairを提供しますが、後戻りを要求すると、あなたの目にタバコを押し付けるように機能しません。解決策は？有料のundoコマンドは必要なく、単に「ロールバック」が実際には何であるかについて、少しずる賢くなる必要があることがわかりました。以下のロールバック実装は、データベースに対して直接SQLを実行することなく、完全にFlywayコマンドを介して達成されます（#minimalism）。仕組みはこうです、お嬢さんたち、しっかり掴まって。 コアコンセプト：ロールバックは変装したマイグレーション これがメンタルブロックの解除です。Flyway Communityは、渡されたバージョン付きマイグレーションを喜んで実行します。そのマイグレーションがテーブルを作成するか削除するかは気にしません、SQLはSQLです。しかし、Flywayが非常に情熱を注いでいる2つのことがあります。そして、このソリューションはそれらを基盤としています。バージョン番号は常に増加しなければならず、スキーマ履歴テーブルは聖書です。したがって、FlywayにV2を元に戻すように依頼する代わりに、V2の逆の処理を行う新しい、より高いバージョンのマイグレーションを作成し、Flywayにそれを実行するように依頼します。基本的に、migrateとして表現された「undo」であり、データベースは元の状態に戻ります。その後、マイグレーションとそのそれぞれの「undo」マイグレーションの記録をflyway_schema_historyテーブルからクリーンアップし、ディスク上のファイルとデータベースの状態が一致するようにします。 ディスク上に必要なもの スコープごとに2つのディレクトリ。意図したマイグレーションと、それらのマイグレーションによって適用されたロジックを元に戻す「down」スクリプトの並列セット。例えば： migrations/ V1__create_customers.sql rollbacks/ V1__create_customers.down.sql 命名規則は重要です。「down」スクリプトは、V<version>__プレフィックスで元のマイグレーションと一致させる必要があります。ロールバックの一部として、downスクリプトのバージョンはインクリメントされるため、Flywayはそれらを適用する新しいマイグレーションとして認識します（これについては後述します）。ロールバック中に何も見つからないことがないように、アップグレードの準備ができているすべてのマイグレーションに対して同等の「down」スクリプトがあることを事前に検証することをお勧めします。 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] 実際に何をロールバックするのか？ 重要な質問です。誰かがロールバックを実行したとき、具体的に何をロールバックしているのでしょうか？最後のマイグレーション？最後のバッチ？最後のmigrateが一度に3つのファイルを適用した場合はどうなりますか？これは、flyway migrateが実行されるたびに正常に適用されたマイグレーションのバッチを記録することで解決されます。これは、migrateの前後で適用されたバージョンをdiffし、適用されたバージョンをJSONファイルに保存することによって行われます。基本的に、新しいバージョンが何であれ、それが「バッチ」です。 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 # `migrate`の前後に適用されたマイグレーションの差分を取得 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を2回実行した場合、2回目の実行は何も新しいものを適用せず、バッチは空になり、記録されたバッチは[]で上書きされません。 ロールバックスクリプトのステージング これで、最後のmigrate呼び出しで適用されたものの記録を永続化できたので、ロールバックするものが正確にわかります。Flywayの「バージョンは常に増加しなければならない」という条件を満たすために、変更されたバージョンプレフィックスを持つ「down」マイグレーションのステージングを開始できます。[1, 2]のようなバッチが与えられた場合、Flywayの消費専用の合成マイグレーションでいっぱいのテンポラリディレクトリを構築します。 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) sql = _down_file(config, version).read_text() (staging / f"V{new_version}__undo_V{version}.sql").write_text(sql) これで、ロールバックで使用されるすべての「down」ファイルが、Flywayが容易に適用できる形式で用意できました。注目すべき点が2つあります。 1. バッチを逆順にします。V1が適用され、次にV2が適用された場合、V2はV1の前に元に戻される必要があります。したがって、最後に適用されたマイグレーションが最初に元に戻され、最も低い新しいバージョン番号が付けられます。現在のハイウォーターマークが2であるバッチ[1, 2]は、次のようになります： V3__undo_V2.sql V4__undo_V1.sql 2. バージョン番号は増加し続けます。1または2を再利用することはありません。Flywayの絶対的なルールは、バージョンは増加する一方であるということです。そのため、undoスクリプトは現在の最大値の上に番号が付けられます。これにより、逆処理を通常のフォワードマイグレートとしてCommunity Editionにこっそり適用できます。 flyway_schema_historyテーブルの整合性を保つ ここで停止し、「down」スクリプトのみをステージングしてmigrateを実行した場合、flyway infoは、これらのマイグレーションがマイグレーションディレクトリにファイルとして存在しないにもかかわらず（そして、これらのマイグレーションがデータベース内の以前のマイグレーションのロジックをキャンセルし、初期状態に戻すにもかかわらず）、V3__undo_V2、V4__undo_V1のレコードを表示します。これらのマイグレーションはファイルとして存在しないため、Flywayは履歴が同期していないとフラグを立て、将来のvalidateおよびmigrateコマンドでエラーを発生させます。そのため、テンポラリディレクトリにステージングする最後のファイルは、「down」スクリプトと共に、履歴テーブルからラウンドトリップの行を削除するafterMigrate.sqlコールバックです。 # 元のマイグレーションとその対応する「down」マイグレーションの両方の行を削除 versions = ", ".join(f"'{v}'" for v in batch + undo_versions) (staging / "afterMigrate.sql").write_text(f"DELETE FROM {config.schema}.flyway_schema_history "f"WHERE version IN ({versions});\n") ラウンドトリップの両側、つまり元のエントリ（1、2）とundoエントリ（3、4）を削除します。すべてが落ち着くと、履歴テーブルはこのすべてが発生したという記憶がなくなります。customersテーブルは削除され、V1__create_customers.sqlはディスク上に残っており、flyway infoはそれを再びPendingとして親切に報告し、あたかもマイグレーションしたことがないかのように再適用する準備ができています。afterMigrateは、バージョン付きマイグレーションが成功した後、Flyway自身の実行内で実行されます。マイグレーションが成功した場合にのみトリガーされます。 アトミックにする ロールバック機能が必要な理由は、物事がうまくいかないからです。したがって、ロールバック自体が、半ば元に戻された状態のままにしておくべきではありません。ここでFlywayフラグに依存します。 def migrate(config: PgConfig, locations: list[str]) -> int: # 単一のDBトランザクションとして実行します。保留中のマイグレーションが失敗した場合、バッチ全体がロールバックされます。 args = _base_args(config, locations) + ["-group=true", "migrate"] return subprocess.run(args).returncode -group=true は、保留中のすべてのマイグレーションを単一のトランザクションにラップします。すべての「down」スクリプトとパージコールバックが一緒にコミットされるか、Postgresがすべてをロールバックし、データベースは変更されません。V2が元に戻されたがV1は元に戻されなかったという状態はありません。（これはDDLがトランザクション可能であることを前提としています）そして、組み立てられたロールバックがすべてをまとめます。 def rollback(config: PgConfig, scope: str) -> int: # 直前に適用されたバッチを読み取ります。