FastAPI + SQLAlchemy 2.0：非同期DB操作という「修羅の道」を攻略する

「なぜ俺のAPIは、リクエストが増えると途端にレスポンスが鈍くなるんだ？」

その答えはシンプルだ。同期的なDBドライバが、I/O待ちのたびにイベントループを窒息させているからだ。FastAPIの真価である「非同期I/O」を最大限に引き出すには、SQLAlchemy 2.0の非同期対応を避けて通れない。しかし、この移行は単なるコードの書き換えではない。同期処理の直感的な思考を捨て、イベントループの挙動を理解するという「設計思想のアップデート」を強いる過酷な道のりだ。

この記事では、僕が幾度となく「RuntimeError」に絶望し、深夜のデバッグでようやく辿り着いた、本番環境で生き残るための非同期DB実装の最適解を共有する。

---

1. なぜ「非同期」が必要なのか？

同期的なDB操作は、SQLを発行してから結果が返ってくるまで、そのスレッドを完全に占有する。リクエストが10件ならまだいい。だが100件、1000件と並行してくると、スレッドプールは枯渇し、APIは悲鳴を上げる。

SQLAlchemy 2.0が非同期対応に舵を切ったのは、この「I/Oブロッキングによるスループット低下」を物理的に解決するためだ。ただし、非同期化は複雑性の増大と引き換えである。これから紹介する実装は、その複雑さを「制御可能な設計」に落とし込むための構成だ。

---

2. 実装：AsyncSessionの正しい構築

`async_sessionmaker` を使った、本番運用を想定した実装例を見てほしい。

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import declarative_base

# 接続プールサイズを明示的に設定。デフォルトのままだと高負荷時に即死する
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    pool_size=20,          # 同時接続のベース数
    max_overflow=10,       # スパイク時に許容する最大接続数
    pool_pre_ping=True     # 接続切断を検知して自動再接続する（必須）
)

AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False, # ここが最重要
)

Base = declarative_base()

async def get_db():
    """リクエストスコープでセッションを管理する"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

なぜ `expire_on_commit=False` なのか？

デフォルトの `True` だと、コミットした瞬間にセッション内の全オブジェクトが「期限切れ」とマークされる。その後で属性にアクセスしようとすると、DBへの再クエリが走り、非同期コンテキスト外で実行されると `DetachedInstanceError` を吐く。 本番環境でこれをやると、「たまに落ちる」という最悪の再現性デバッグに追い込まれる。 潔く `False` にし、必要なら明示的に `session.refresh()` を呼ぶのが鉄則だ。

---

3. Alembic移行：迷宮からの脱出

Alembicの非同期マイグレーションは、`env.py` の設定がすべてだ。

async def run_async_migrations():
    # 非同期エンジンを構築
    connectable = create_async_engine(config.get_main_option("sqlalchemy.url"))

    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)

    await connectable.dispose()

def do_run_migrations(connection):
    # 同期的なマイグレーション実行をラップする
    context.configure(connection=connection, target_metadata=target_metadata)
    with context.begin_transaction():
        context.run_migrations()

ここで `postgresql+asyncpg://` を忘れれば、接続エラーの無限ループが待っている。マイグレーションはアプリの土台だ。ここだけは疎かにしてはいけない。

---

4. 現場で直面する「壁」と解決策

① 同期ライブラリとの共存

「どうしても同期処理のライブラリを使いたい」という場面は必ず来る。その場合は、`anyio` や `loop.run_in_executor` を使って、メインのイベントループを止めずに別スレッドで実行させるのが正攻法だ。

import asyncio
from functools import partial

# イベントループをブロックせずに同期関数を実行
result = await asyncio.get_running_loop().run_in_executor(
    None, partial(some_sync_function, arg1)
)

② セッションの使い回しは「死」

`Depends` で取得したセッションを、別タスクやバックグラウンド処理に渡してはいけない。セッションはあくまで「そのリクエストの寿命」に従うべきだ。バックグラウンドでDBを触るなら、必ずそのタスク内で別途セッションを生成せよ。

---

最後に：沼を楽しめるか？

SQLAlchemy 2.0の非同期対応は、正直言ってまだ荒削りだ。トレースバックは複雑さを増し、少しのミスが致命的なリークを招く。しかし、非同期で駆け抜けるAPIのスループットを目の当たりにすれば、その苦労が「無駄ではなかった」と確信できるはずだ。

技術は常に「楽をするため」ではなく「より高い要求に応えるため」に進化する。この沼にハマり、泥をすすりながら実装する経験こそが、君をただのエンジニアから、圧倒的な成果を出せるエンジニアへと変える。

さあ、コーヒーを淹れ直して、もう一度コードを書き始めよう。その一行が、爆速APIの礎になる。

---