SQLAlchemy 2.0 #7 Alembic と実戦構成 — マイグレーション、非同期、チーム規約

シリーズをここまで追ってきた方は、モデルを宣言し、セッションで操作し、リレーションとクエリを扱えるようになっています。残っているのは運用です。スキーマは必ず変わり、トラフィックが増えれば非同期が必要になり、チームが大きくなれば規約が必要になります。最終回はこの 3 つです。

create_all の限界と Alembic #

Base.metadata.create_all(engine) は存在しないテーブルを作るだけです。既存のテーブルにカラムを足すことも、型を変えることも、インデックスを作ることもしてくれません。モデルと実際の DB がずれ始めた瞬間から、スキーマ変更の履歴を管理する道具が必要になります。SQLAlchemy 陣営の標準が、同じ開発者の作った Alembic です。

uv add alembic
alembic init migrations

alembic init が作るもののうち、手を入れる場所は 2 か所です。

• alembic.ini: sqlalchemy.url に DB の接続 URL を設定します(実務では環境変数から読むよう env.py 側で処理するほうがよいです)。
• migrations/env.py: target_metadata = Base.metadata でモデルのメタデータをつなぎます。autogenerate が比較する「目標の状態」がこれです。

以降のサイクルは 3 つのコマンドの繰り返しです。

# 1. モデルを修正したら、現在の DB との差分でマイグレーションファイルを生成
alembic revision --autogenerate -m "user に nickname カラムを追加"

# 2. 生成されたファイルを開いてレビュー(必須!)

# 3. 適用
alembic upgrade head

autogenerate はあくまで下書きです #

autogenerate は Base.metadata(目標)と実際の DB(現在)を比較し、upgrade と downgrade の関数が埋まった Python ファイルを生成します。よく検知できるものと、できないものを知っておく必要があります。

いちばん危険なのが名前の変更です。nickname を alias に変えると、autogenerate は「nickname 削除、alias 追加」を生成し、そのまま適用するとデータが消えます。 ファイルを開いて op.drop_column + op.add_column を op.alter_column(..., new_column_name=...) に書き換えなければなりません。「autogenerate の結果は下書きであり、レビューなしでは適用しない」をチームのルールにしておく理由です。

本番適用ではさらに 2 つ守ります。第一に、マイグレーションファイルはコードと同じ PR でレビューします。第二に、大きなテーブルへのカラム追加やインデックス作成はロック時間を確認します。PostgreSQL の CREATE INDEX CONCURRENTLY のように DB ごとの無停止オプションが必要な場合があり、こういう部分は autogenerate が勝手にやってはくれません。

第 3 回で設定した命名規約がここで効いてきます。制約名が予測可能なので、autogenerate が生成する op.drop_constraint("uq_user_account_email", ...) がどの DB でもそのまま動きます。

非同期 — create_async_engine と AsyncSession #

FastAPI のような非同期フレームワークで同期版の SQLAlchemy をそのまま使うと、DB の待ち時間の間イベントループが塞がります。2.0 は asyncio を正式サポートしており、ここまで学んだ API とほとんど同じ形です。

uv add "sqlalchemy[asyncio]" asyncpg  # PostgreSQL の非同期ドライバー

from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

engine = create_async_engine("postgresql+asyncpg://user:pw@localhost/mydb")
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def list_users() -> list[User]:
    async with AsyncSessionLocal() as session:
        result = await session.scalars(select(User))
        return list(result.all())

変わるところは規則的です。エンジンは create_async_engine、ドライバーは非同期用(asyncpg など)、セッションは AsyncSession、そして I/O が起きる呼び出し(execute、scalars、commit、flush)に await が付きます。モデルの宣言と select() の組み立ては同期版とまったく同じです。

ひとつだけ注意が必要な違いがあります。非同期セッションでは遅延ロードがデフォルトでは動きません。 user.addresses へのアクセスが暗黙に I/O を起こすという構造が、await なしには成立しないからです(MissingGreenlet エラーとして現れます)。非同期では第 5 回の eager loading(selectinload)が、選択肢ではなく事実上の必須になります。上の例の expire_on_commit=False も同じ文脈で、コミット後の属性アクセスが再取得を起こさないようにする非同期の慣例です。

プロジェクト構成とチーム規約 #

シリーズの締めくくりに、実戦プロジェクトで検証されてきた構成を整理します。

myapp/
├── app/
│   ├── db.py          # エンジン、sessionmaker(プロセスごとに 1 回生成)
│   ├── models/        # ドメイン別のモデルモジュール、Base は 1 か所に
│   ├── repositories/  # クエリを集める層(任意)
│   └── ...
├── migrations/        # alembic init の生成物
├── alembic.ini
└── pyproject.toml

• エンジンはプロセスごとに 1 つです。リクエストのたびに create_engine を呼ぶのは、プールを毎回作り直しているのと同じです。db.py のようなモジュールで 1 回だけ作り、インポートして使います。
• モデルはドメイン別に分割しても Base は 1 つです。Alembic の target_metadata がすべてのモデルを見る必要があるので、env.py からモデルモジュールがすべてインポートされるようにします。「新しいモデルを作ったのに autogenerate が見てくれない」の原因は、たいていインポート漏れです。
• クエリを集める層(repository でも service でも名前は自由)を置くと、select() の組み立てがビューのコードに散らばらず、N+1 対応(どの取得にどの eager loading を使うか)を 1 か所で管理できます。
• チーム規約 3 つ: スキーマ変更は必ず Alembic を経由する(本番 DB への手動 DDL 禁止)、autogenerate の結果はレビューしてからマージする、セッションのスコープは作業単位ごとに 1 つを守る。この 3 つが守られるだけで、SQLAlchemy まわりの事故のほとんどは防げます。

シリーズを終えて #

7 回を要約するとこうなります。SQLAlchemy は Core の上に ORM が載った 2 層構造で(第 1 回)、エンジンと明示的なトランザクションが土台にあり(第 2 回)、モデルは Mapped で宣言し(第 3 回)、セッションが変更をためて SQL として送り出し(第 4 回)、リレーションは便利だが N+1 を警戒すべきで(第 5 回)、クエリは select() ひとつで組み立てられ(第 6 回)、運用は Alembic と規約が支えます(第 7 回)。FastAPI との連携の形はモダン Python 実践 #3 で、パッケージと依存関係の管理は Python パッケージングシリーズで続きをどうぞ。

まとめ #

• create_all は新しいテーブルを作るだけです。スキーマ変更の履歴管理は Alembic の仕事で、サイクルは revision –autogenerate、レビュー、upgrade head の繰り返しです。
• autogenerate は下書きです。特に名前の変更は削除 + 追加として生成されてデータを失わせるので、レビューで alter_column に直します。
• 非同期は create_async_engine + AsyncSession に規則的に移行できますが、遅延ロードが動かないため eager loading が必須になります。
• エンジンはプロセスごとに 1 つ、Base はプロジェクトに 1 つ、スキーマ変更は常に Alembic 経由。この規約が運用事故を防ぎます。