alembic

alembic — SKILL.md

バリアント： 標準 · 使用するタイミング： スキルが呼び出され、接続済みの Alembic 設定とレビュー済みのマイグレーションスクリプト（または該当する部分）を生成した後、呼び出し元に制御を戻します。

概要

このスキルでは、既存の SQLAlchemy 2.x データレイヤー上で、SQLite、PostgreSQL、MySQL のスキーママイグレーションを1つのマイグレーション履歴で実行する方法をエージェントに説明します。すでに動作する Base.metadata が存在すること（それを用意するのは関連する sqlalchemy スキルの役割です）を前提とし、Alembic をそこに接続します。基本となるのは同期の方法です。同期接続を使った標準的な run_migrations_online() の形を示し、非同期については短く補足します。重要な点は2つあります。（1）自動生成は下書きにすぎず、レビューが必要です。自動生成には記録された見落としがあり、確認せずに適用すると気づかないままデータを失うおそれがあります。（2）SQLite のバッチ／移動・コピーに関する注意点です。SQLite ではほとんどの操作で ALTER を実行できないため、マイグレーションは batch_alter_table 形式で記述します。SQLite ではテーブルを再作成し、PG／MySQL では通常の ALTER を出力します。このスキルでは、create_all() と Alembic のどちらを使うかの判断と、すでに作成済みのデータベースに Alembic を導入するためのベースラインのスタンプ設定も扱います。

使用するタイミング

• ✅ 既存の SQLAlchemy Base.metadata に Alembic（env.py + alembic.ini）を接続し、自動生成で稼働中のスキーマとモデルを差分比較できるようにする場合。
• ✅ マイグレーションを作成する場合（空の revision または revision --autogenerate で下書きを生成）。適用前にレビューして編集します。
• ✅ upgrade／downgrade の実行、history／current／heads の確認、または merge による複数のヘッドの解決を行う場合。
• ✅ SQLite、PostgreSQL、MySQL で適用する必要があるマイグレーションを1つ作成する場合。バッチ（移動・コピー）操作を使用します。
• ✅ プロジェクトで Base.metadata.create_all() と Alembic のどちらを使うか判断する場合、または create_all で作成済みのデータベースに Alembic を導入するためにベースラインをスタンプする場合。

次の場合は有効にしないでください：

• SQLAlchemy モデル／Engine／接続 URL／セッションの定義 → sqlalchemy。このスキルの役割は、すでに存在する Base.metadata に Alembic を接続することだけです。
• マイグレーション対象のスキーマ上にジョブキュー／スケジューラー／リースループを構築する場合 → sql-job-queue。
• バージョン履歴を持たずに SQL DDL を手書きする場合。Alembic はその履歴を管理するためにあります。本当に履歴が不要な場合は、Step 6 の create_all() に関する判断を参照してください。

ワークフロー

Step 1: マイグレーション環境を初期化する

alembic init を実行すると、プロジェクトの隣に必要な構成が作成されます。

alembic init alembic        # creates alembic.ini + alembic/ (env.py, script.py.mako, versions/)

これにより、alembic.ini（設定）、alembic/env.py（実行時フック）、alembic/script.py.mako（リビジョンのテンプレート）、alembic/versions/（リビジョンファイルの格納先）が作成されます。alembic.ini の script_location は alembic/ ディレクトリを指します。version_locations（任意）を使うと、複数のディレクトリにリビジョンを分けられます。

Step 2: DB URL は環境変数から取得する（ハードコードしない）

生成された alembic.ini には、sqlalchemy.url = driver://... という値が直接記述されています。実際の URL をここにコミットしないでください。空欄のままにして env.py で環境変数から設定するか、補間してください。env.py で設定すれば、アプリと共有する情報源を1つにできます。

# alembic/env.py
import os
from alembic import context

config = context.config
config.set_main_option("sqlalchemy.url", os.environ["DB_URL"])

set_main_option は実行時に alembic.ini の値を上書きします。そのため、DB_URL を変更するだけで、同じマイグレーション履歴をローカルでは SQLite、ほかの環境では PostgreSQL／MySQL に対して実行できます。

Step 3: target_metadata をモデルの Base.metadata に設定する

これが --autogenerate を機能させるための接続部分です。指定した MetaData と稼働中の DB を差分比較します。モデルの Base をインポートし（すべてのテーブルが metadata に登録されるようにします）、次のように割り当てます。

# alembic/env.py (continued)
from myapp.models import Base   # your DeclarativeBase subclass — sqlalchemy sibling owns this

target_metadata = Base.metadata

モデルを定義しているモジュールをインポートする必要があります。インポートされていないテーブルは target_metadata から見えないため、自動生成でも検出されません（見落としの一例です。Step 5 を参照）。

Step 4: env.py のオンラインランナー（同期優先）にバッチモードと命名規則を設定する

スキャフォールドにはオフラインランナーとオンラインランナーが用意されています。オンラインではライブ接続を使用します（一般的な方法です）。オフラインでは、DB に接続せず SQL を標準出力またはファイルに出力します（--sql モード）。これは、手動で適用するデプロイや、レビュー承認を経て適用するデプロイに使用します。SQLite のマイグレーションを機能させるため（Step 7）、オンラインランナーを編集してバッチモードを有効にし、バッチテーブル再作成で制約を名前で参照できるように naming_convention を定義します。

# alembic/env.py (continued)
from sqlalchemy import engine_from_config, pool, MetaData

NAMING_CONVENTION = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}
# Best applied on Base.metadata itself in the models module so live + autogen agree;
# shown here for visibility. If set in models: MetaData(naming_convention=NAMING_CONVENTION)

def run_migrations_online() -> None:
    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            render_as_batch=True,          # makes SQLite migrations move-and-copy (Step 7)
            compare_type=True,             # also diff column TYPE changes (still review them)
            compare_server_default=True,   # attempt server_default diffs (imperfect — Step 5)
        )
        with context.begin_transaction():
            context.run_migrations()

run_migrations_online()   # the scaffold guards this with context.is_offline_mode()

render_as_batch=True は、生成される ALTER 文をバッチモードで囲むグローバルスイッチです。短時間で終了するマイグレーションプロセスには NullPool が適しています。（スキャフォールドのオフラインランナーである run_migrations_offline() は context.configure(url=..., literal_binds=True) を呼び出して SQL を出力します。--sql を使ったデプロイ用に、この設定は残してください。）

Step 5: リビジョンを作成する — autogenerate はドラフトを生成し、レビューは自分で行う

リビジョンを作成する方法は 2 つあります。

alembic revision -m "add jobs table"              # EMPTY — you write upgrade()/downgrade() by hand
alembic revision --autogenerate -m "add jobs table"   # DRAFTS ops by diffing target_metadata vs DB

autogenerate はドラフト生成ツールであり、正解を示すものではありません。target_metadata とライブ DB を比較して、最善の推測をファイルに書き込みますが、既知の検出漏れがあります。生成されたファイルは必ず開いて確認・編集してから適用してください。 autogenerate では、次の項目を確実に検出できません。

• テーブル名やカラム名の変更 — 古いものの削除と新しいものの追加として認識されます。何も確認せずに適用すると、データを失います。op.rename_table(...) またはバッチモードでのカラム名変更を手動で記述してください。
• 多くの server_default の変更 — compare_server_default=True を設定しても、見落とされたり、誤った形式で出力されたりするものが多数あります。
• CHECK 制約および一部の名前付き制約 — 頻繁に検出されません。
• 一部のインデックス変更とカラムの型変更 — compare_type=True は役立ちますが、すべてを検出できるわけではありません。出力された型を確認してください。
• target_metadata から参照できないものすべて — インポートし忘れたモデルや、Alembic に比較対象として指定されていないスキーマオブジェクト（他のスキーマや特定のオブジェクト型など）。見えないものは差分にも現れません。

レビュー済みの autogenerate ファイルは次のようになります。明示的に記述された、元に戻せる downgrade() に注目してください。

"""add jobs table"""
from alembic import op
import sqlalchemy as sa

revision = "a1b2c3d4e5f6"
down_revision = None        # first migration; otherwise the prior revision id
branch_labels = None
depends_on = None

def upgrade() -> None:
    op.create_table(
        "jobs",
        sa.Column("id", sa.Integer(), primary_key=True),
        sa.Column("name", sa.String(length=200), nullable=False),
        sa.Column("data", sa.JSON(), nullable=True),
    )
    op.create_index("ix_jobs_name", "jobs", ["name"])

def downgrade() -> None:
    op.drop_index("ix_jobs_name", table_name="jobs")
    op.drop_table("jobs")

downgrade() は upgrade() の正確な逆になるように記述します。操作の順序を逆にし、それぞれの操作も逆にしてください。downgrade が誤っていたり pass になっていたりするマイグレーションは、ロールバックできません。安全に名前を変更する方法とデータマイグレーションについては、references/alembic-extras.md を参照してください。

Step 6: 適用／ロールバック／確認

alembic upgrade head        # apply every unapplied revision up to the latest
alembic upgrade +1          # apply the next one only
alembic downgrade -1        # roll back the most recent revision
alembic downgrade base      # roll back everything (empty schema)

alembic current             # which revision the DB is stamped at
alembic history             # the revision graph, newest first
alembic heads               # the current head(s) — more than one means a branch to merge

Step 7: 複数の方言に対応する — バッチ形式でマイグレーションを記述する

これは、複数の方言に対応するうえで最も重要なルールです。SQLite ではほとんどのスキーマ要素を ALTER できません。通常の ALTER TABLE では、カラムの削除、カラムの型や NULL 許容性の変更、ほとんどの制約の追加ができません。Alembic のバッチ操作では、移動とコピーによってこれを回避します。対象のスキーマで新しいテーブルを作成し、INSERT ... SELECT で行を移し、古いテーブルを削除して、新しいテーブルの名前を変更し、元の場所に配置します。PostgreSQL と MySQL では、まったく同じバッチコードから通常の ALTER 文が生成されます。そのため、すべての変更をバッチ形式で記述すれば、3 つすべての方言で適用できる 1 つのマイグレーションスクリプトを作成できます。

def upgrade() -> None:
    # Same code path on all three; SQLite does move-and-copy, PG/MySQL do ALTER.
    with op.batch_alter_table("jobs") as batch_op:
        batch_op.add_column(sa.Column("priority", sa.Integer(), nullable=False, server_default="0"))
        batch_op.alter_column("name", existing_type=sa.String(200), nullable=True)
        batch_op.create_unique_constraint("uq_jobs_name", ["name"])

def downgrade() -> None:
    with op.batch_alter_table("jobs") as batch_op:
        batch_op.drop_constraint("uq_jobs_name", type_="unique")
        batch_op.alter_column("name", existing_type=sa.String(200), nullable=False)
        batch_op.drop_column("priority")

render_as_batch=True（Step 4）を設定すると、autogenerate によって batch_alter_table ブロックが生成されます。naming_convention（Step 4）を設定すると、SQLite の移動とコピーの際に制約を維持したままテーブルを再作成できます。名前のない制約があると、再作成時に削除される可能性があります。生成された SQL を確認する際、SQLite では CREATE TABLE _alembic_tmp_... / INSERT ... SELECT / DROP / ALTER ... RENAME の順に実行され、PG/MySQL では単一の ALTER TABLE が生成されることを想定してください。リフレクションだけでは不十分な場合の batch_alter_table(..., copy_from=...) や、再作成モードの制御については、references/alembic-extras.md を参照してください。

Step 8: ブランチ分岐 — 複数の head を merge で解決する

2 人の開発者（または 2 つの機能ブランチ）が同じ親からそれぞれリビジョンを追加すると、alembic heads には2 つの head が表示され、upgrade head はエラーになります（曖昧なためです）。マージリビジョンでつなぎ合わせます。マージリビジョンは両方を親に持ち、自身ではスキーマを変更しません。

alembic heads                              # shows e.g. two head ids: rev_a, rev_b
alembic merge -m "merge a and b" rev_a rev_b
alembic upgrade head                       # now unambiguous

ステップ 9：非同期についての補足（短縮版）

非同期の SQLAlchemy スタックでも、マイグレーションは一般に同期で実行します。最も簡単な方法は、Alembic に同期ドライバーの URL を指定し、ステップ 4 のランナーを変更せずに使うことです。非同期エンジンを使う必要がある場合は、env.py のオンラインランナーを次のようにします。

import asyncio
from sqlalchemy.ext.asyncio import async_engine_from_config

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata, render_as_batch=True)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}), prefix="sqlalchemy.", poolclass=pool.NullPool
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)   # greenlet bridge to Alembic's sync context
    await connectable.dispose()

asyncio.run(run_async_migrations())

run_sync(do_run_migrations) は、グリーンレットブリッジを介して、非同期接続から Alembic の同期マイグレーションコンテキストに移行します。非同期固有の部分はこれだけです。それ以外（リビジョン、バッチ、レビュー）はすべて同じです。

ルール

必ず守るルール（例外なし）：

• 自動生成はドラフトとして扱い、適用前に生成されたスクリプトをすべて確認・編集してください。 レビューしていない自動生成リビジョンを upgrade してはいけません。リネームや制約の変更によって、データが気づかないうちに失われます。
• SQLite を対象にする場合は、バッチ形式でマイグレーションを記述してください。 SQLite が通常の ALTER として実行できない変更には、render_as_batch=True と op.batch_alter_table(...) を使います。同じコードを PG／MySQL でも正しく使えます。
• target_metadata = Base.metadata を設定し、モデルをインポートしてください。 こうすることで自動生成がすべてのテーブルを認識できます。インポートされていないテーブルは認識されません。
• DB URL は、コミットされた alembic.ini のリテラルではなく、環境変数から取得してください。 履歴は 1 つのまま、複数の方言や環境に対応できます。
• すべての upgrade() に、正確に逆操作となる downgrade() を用意してください。 元に戻せる操作の downgrade を pass にしてはいけません。
• naming_convention を使ってください。 SQLite でのバッチによる移動・コピー時に、名前付き制約を参照して保持できます。

推奨事項（必要に応じて変更可能）：

• 同期ランナーを優先してください。同期ドライバーの URL が実際に利用できない場合に限り、非同期の run_async_migrations 形式を使います。
• リビジョンのひな型には --autogenerate を使い、その後手作業で編集する方法を優先してください。操作を最初から記述する方法は、純粋なデータマイグレーション（これは手作業で記述します）を除き、避けてください。
• 差分の精度を高めるため、compare_type=True／compare_server_default=True を有効にしておきます。ただし、その出力は保証ではなくドラフトとして扱ってください。
• マイグレーションプロセスには NullPool を使ってください。プロセスの実行時間は短く、コネクションプールを保持する必要はありません。

注意点

• 自動生成では、リネームが削除＋追加として検出され、データが失われます。 レビューで最も重要な確認事項です。列やテーブルの名前を変更した場合、生成された drop_column＋add_column（または drop_table＋create_table）はデータを破棄します。op.rename_table(...) またはバッチの alter_column(..., new_column_name=...) に書き換えてください。
• 新しいモデルをインポートしないと、自動生成の結果が空になります。 テーブルが target_metadata に登録されていないため、自動生成は「変更なし」と判断します。env.py 内またはその前に、モデルモジュールをインポートしてください。
• バッチを使わないと、SQLite で ALTER が失敗します。 一部の制約を伴う add_column、および drop_column／alter_column は、ほとんどの add_constraint と同様に、batch_alter_table でラップしないと SQLite でエラーになります。このエラーは、開発に使った PG ではなく、SQLite でマイグレーションを実行したときに初めて発生することがよくあります。最初から render_as_batch=True を有効にしてください。
• SQLite では、名前のない制約はバッチ再作成時に失われます。 移動・コピーによる再作成ではテーブルが作り直されるため、Alembic が名前を付けられない制約（naming_convention がない場合）は保持されないことがあります。失っては困るデータができる前に、規約を定義してください。
• alembic upgrade head で「multiple heads」エラーが発生します。 これは、マージされていないブランチが 2 つある状態です。alembic heads で確認し、alembic merge で統合してください（ステップ 8）。破損ではなく、ブランチが未解決なだけです。
• server_default の差分は信頼できません。 compare_server_default=True を指定していても、デフォルト値の変更が見落とされたり、誤って出力されたりすることがよくあります。デフォルト値は手作業で確認し、自動生成の結果を当てにしないでください。
• downgrade base はスキーマを消去します。 すべての downgrade() を実行し、空の状態にします。開発環境では問題ありませんが、本番環境では破壊的です。実行前に、DB_URL がどの DB を指しているか確認してください。

アンチパターン

• 「自動生成はたいてい正しい」からといって、自動生成されたマイグレーションを読まずに適用しないでください。自動生成の盲点（名前変更、サーバーデフォルト、CHECK 制約／名前付き制約）こそ、データを破壊するケースです。
• 「Postgres では動くから」といって、単純な op.alter_column / op.drop_column を書き、どこでも実行できると思い込まないでください。SQLite ではエラーになります。batch_alter_table を使ってください。
• 「そのほうが簡単だから」といって、Base.metadata.create_all() をマイグレーションの仕組みとして出荷しないでください。バージョン履歴も、段階的な変更も、ダウングレードもありません。以下の判断を参照してください。スキーマを一度出荷したら、Alembic を使います。
• 「単なる設定だから」といって、本番 URL を alembic.ini にハードコードしないでください。認証情報が漏えいし、特定の方言に固定されます。環境から取得してください。
• 「ロールバックすることはないから」といって、downgrade() を pass のままにしないでください。元に戻せないマイグレーションがあると、それ以降のすべてのロールバックがそこで阻まれます。
• down_revision を編集して、2 つの head を手動でマージしないでください。両方の親を正しく記録する alembic merge を使ってください。

create_all() と Alembic の使い分け（このスキルが判断します）

Base.metadata.create_all(engine) は、現在のモデル状態に対する CREATE TABLE を一度に出力します。これは、グリーンフィールド／未出荷のスキーマ、使い捨てのデータベース、テスト／開発用の初期化（sqlalchemy の関連スキルも、まさにこの用途で使用しています）に限って適切です。出荷済みのものに使うには致命的な制限があります。バージョン履歴も、段階的な変更も、ダウングレードもなく、最初から作成することしかできず、決して進化させられません。

スキーマを出荷したとき、複数の環境で実行するとき、または時間とともに進化させるとき（チームが順番に適用する変更の DAG）には、Alembic を使ってください。マイグレーション履歴は、そのためにあります。

create_all() で構築したデータベースに Alembic を導入する場合 — 作成を再実行しないでください。代わりに、次の手順で行います。

1. Alembic を初期化し（手順 1～4）、現在のスキーマを作成する upgrade() を含むベースラインマイグレーションを作成します（空の DB に対して自動生成するか、create_all が生成したものに合わせて手動で記述します）。
2. 既存の create_all 済み DB で alembic stamp head を実行します。これにより、実行せずに（CREATE TABLE は再実行されません）、DB がベースラインのリビジョンに到達済みであることを記録します。
3. 以降、すべての変更は新しいリビジョンとして作成し、upgrade で適用します。

alembic stamp <rev> は、リビジョンの本体を実行せずに、適用済みとして記録するため alembic_version 行を書き込みます。これは、管理されていない create_all 済み DB から、管理された履歴へ移行するための橋渡しです。

出力

設定済みの Alembic 構成（alembic.ini と、target_metadata = Base.metadata、環境から取得する DB URL、render_as_batch=True、naming_convention を含む env.py）に加え、レビュー済みで元に戻せるリビジョンファイルを versions/ 以下に作成します。SQLite、PostgreSQL、MySQL で同じ履歴を適用できるよう、バッチ形式で記述します。この成果物は次のレイヤーで利用されます。alembic upgrade head を実行するデプロイ手順、または移行済みスキーマを基盤とする上位のパターン（例：sql-job-queue）です。

関連項目

• sqlalchemy — このスキルの基盤となるデータレイヤーです。DeclarativeBase／Mapped モデル、Engine／URL、セッション、方言間での ORM の注意点を扱います。このスキルは Base.metadata が存在することを前提とし、そこに Alembic を接続するだけです。sqlalchemy は開発／テスト用の初期化に create_all() を使い、管理された変更を行うかどうかの判断はこのスキルに委ねます。
• sql-job-queue — 移行済みスキーマを基盤とする、ジョブストア／スケジューラのパターン（ready-set クエリ、アトミックなリース、公平な割り当て）です。このスキルはスキーマ履歴を作成しますが、キューについては説明しません。
• pydantic-v2 — アプリケーション境界での検証を扱います（DB スキーマやマイグレーションではありません）。

段階的な開示

• references/alembic-extras.md — より詳しいマイグレーション例です。データを保持した安全な名前変更、op.bulk_insert / op.execute によるデータマイグレーション、リフレクションだけでは不十分な場合の batch_alter_table(..., copy_from=..., recreate=...)、オフラインの --sql モード、autogenerate の対象を絞る include_object / include_name を扱っています。読み込む場面： 本文のスニペットだけでは、特定のマイグレーション形式への対応に不十分な場合。
• references/sources.md — このスキルの調査に関する出典情報です。読み込む場面： 主張の出典を確認するとき、または新たにレビューするとき。

本文の上限

• description は 1,024 文字以下。
• 本文は約 500 行 / 5,000 トークン以下。
• 参照ファイルごとに、10k トークン超で警告、25k トークン超でエラー。参照ファイルの合計は、25k トークン超で警告、50k トークン超でエラー。