FastAPI + SQLAlchemy 2.0：非同期DBの「謎エラー」を撲滅する最短ルート

FastAPIで非同期DBを触り始めた初心者が、3日以内に必ず直面する「謎のクラッシュ」。Googleで検索しても出てくるのは、古臭いチュートリアルのコピペと、複雑怪奇なラッパーライブラリの宣伝ばかりだ。

正直に言おう。SQLAlchemy 2.0の非同期対応に、複雑な設定なんて1ミリもいらない。みんながこぞって書いている長ったらしい解説記事は、今すぐゴミ箱に捨てていい。本質は驚くほどシンプルだ。

1. 理想的なディレクトリ構成

まず、沼にハマらないための地図を渡しておく。迷走する奴の9割は、ファイル構成が破綻している。

.
├── alembic/            # マイグレーション管理
├── app/
│   ├── database.py     # 接続設定とセッション管理
│   ├── models.py       # データベースモデル（Baseの定義含む）
│   └── main.py         # FastAPIエントリーポイント
├── alembic.ini
└── .env

2. 「脱・初心者」のDB接続設定

`AsyncSession`をいちいち関数でラップして……なんてやってるとコードが汚れるし、管理が地獄になる。`async_sessionmaker`を`Depends`で注入する。これだけで十分だ。

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False, # これが命綱
)

class Base(DeclarativeBase):
    pass

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

なぜ `expire_on_commit=False` が必要なのか？

これがないと、コミットした瞬間にオブジェクトの属性がDBから強制的にクリアされる。その結果、レスポンスを返す前に`DetachedInstanceError`で爆死する。 副作用の注意点: これをオフにすると、セッション内で取得したデータが古い状態のまま保持される可能性がある。複雑なトランザクションを回す際は、適宜 `session.refresh(obj)` を呼ぶ癖をつけろ。

3. Alembicで沼らないための「env.py」修正術

「マイグレーションが動かない」と嘆く奴のほとんどは、`alembic/env.py`の`target_metadata`を設定していない。ここが空だと、Alembicはモデルの存在を認識できない。

# alembic/env.py
from app.database import Base
from app.models import User # ここに全モデルをインポートする

target_metadata = Base.metadata

モデルを追加するたびにここに追記するのは手間だが、これが最も確実だ。もし「ファイルが増えて面倒」なら、`models/__init__.py`で全モデルをインポートし、`Base.metadata`をそこから参照させる設計にしろ。

4. なぜ「モデル集約」がベストなのか

`database.py`と`models.py`を分けると、必ずと言っていいほど「循環インポート」という悪魔が顔を出す。

俺の結論は、`models.py`に`Base`を定義し、それを各所からインポートする構成が最強だ。`database.py`は接続の裏方役に徹させる。これで依存関係のループを物理的に断ち切れる。

5. SQLAlchemy 2.0記法：型安全は正義

`db.Column`なんて書いているのは、もはや遺物だ。`Mapped`と`mapped_column`を使え。

from sqlalchemy.orm import Mapped, mapped_column
from .database import Base

class User(Base):
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(unique=True)

IDEの補完が完璧に効く。型定義がそのままDBスキーマになる。実行時エラーが激減する。これを使わない理由が見当たらない。

6. 最後に：これだけはやれ（チェックリスト）

1. `await`の忘却を疑え: 非同期処理で「結果が空」なら、`await`を忘れてコルーチンオブジェクトをそのまま叩いている可能性が高い。デバッグ時は`print(result)`ではなく、`print(type(result))`で型を確認しろ。
2. `expire_on_commit=False`の徹底: `DetachedInstanceError`に遭遇したら、まずはここを見直せ。
3. `target_metadata`の確認: マイグレーションでテーブルが生成されない時は、`env.py`にモデルが正しくインポートされているか、`alembic.ini`のURLが正しいか、この2点だけを執拗に確認しろ。

ネット上の「ベストプラクティス」を鵜呑みにせず、一度自分の手で`DetachedInstanceError`に絶望してみる。それが一番の近道だ。複雑なライブラリに逃げるな。SQLAlchemy公式の「Mapped記法」と「AsyncSession」をシンプルに使いこなせば、それだけで今のバックエンド開発は十分戦える。

---