📌
🚀 AsyncSession × SQLModel × Pytest 実践ガイド：非同期DBテストの正しい作り方

2026/01/31に公開
 🌐 はじめにFastAPIを本格的に使い始めると、async def による非同期エンドポイントと AsyncSession（非同期DBセッション）を組み合わせた構成が主流になります。これは、高い同時処理性能とレスポンス性能を実現できる一方で、設計やテストの難易度を大きく引き上げる要因にもなります。

特に、非同期DBを使ったAPIのテストでは、同期版では意識しなくてよかった問題が顕在化します。
AsyncSession の初期化方法が分かりにくい
テスト実行時にイベントループが衝突する
DBの初期化やクリーンアップの責務が曖昧になる
CI環境でだけテストが失敗する
本記事では、AsyncSession × SQLModel × Pytest を組み合わせたときの「壊れにくく、再現性の高い正しい構成」を、実務でそのまま流用できる形で体系的に解説します。

 🧩 使用技術と前提本記事で扱う技術スタックは以下の通りです。
FastAPI：非同期処理を前提としたモダンWebフレームワーク
SQLModel：PydanticとSQLAlchemyを統合したORM
AsyncSession：SQLAlchemyの非同期セッション
Pytest：Python標準のテストフレームワーク
pytest-asyncio / anyio：非同期テストを安定させるための補助ライブラリ
SQLite（aiosqlite）：テスト用の軽量な非同期DB
前提として、以下の内容を理解していることを想定しています。
FastAPIでの基本的なAPI作成経験
Pytestでの基本的なテスト記述
SQLModel（同期版）の利用経験
同期DBテストとの違いを意識しながら読み進めると理解が深まります。

 🗂️ ディレクトリ構成まずは、非同期DBテストを前提としたディレクトリ構成を確認します。
project/
├── app/
│   ├── main.py
│   ├── database.py
│   ├── models.py
│   └── routers/
│       └── items.py
├── tests/
│   ├── conftest.py
│   └── test_items_async.py
└── requirements.txt
この構成のポイントは、
DB接続ロジックを database.py に完全に分離
FastAPI依存関係として Session を注入
tests 側から依存関係を差し替え可能
という点にあります。これにより、本番コードを変更せずにテスト環境だけを柔軟に構築できます。

 🛢️ SQLModel + AsyncSession のDB設定次に、AsyncSessionを使ったDB設定を行います。
app/database.py
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite+aiosqlite:///./app.db"

engine = create_async_engine(DATABASE_URL, echo=True)

AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_session() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session
expire_on_commit=False を指定することで、commit後もオブジェクトが無効化されず、APIレスポンスでそのまま利用できます。これはテストだけでなく、本番コードでも重要な設定です。

 🧪 AsyncSession対応のモデル定義モデル定義自体は同期版とほぼ同じですが、非同期環境でも問題なく使える点がSQLModelの大きな利点です。
app/models.py
from sqlmodel import SQLModel, Field

class Item(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    price: int
このモデルは、
APIリクエストボディ
DBテーブル定義
テストデータ構造
のすべてを1つで兼ねるため、非同期テストでも扱いやすくなります。

 🔄 非同期CRUDエンドポイント次に、AsyncSessionを使ったCRUDエンドポイントを定義します。
app/routers/items.py
from fastapi import APIRouter, Depends
from sqlmodel.ext.asyncio.session import AsyncSession
from app.database import get_session
from app.models import Item

router = APIRouter()

@router.post("/items/")
async def create_item(item: Item, session: AsyncSession = Depends(get_session)):
    session.add(item)
    await session.commit()
    await session.refresh(item)
    return item
非同期CRUDでは、
await session.commit()
await session.refresh()
を忘れると、テスト時に意図しない挙動を引き起こすため注意が必要です。

 🧪 テスト用Async DBをfixtureで構築ここが非同期DBテストの最重要ポイントです。テスト専用のDBとイベントループを fixture で完全に管理します。
tests/conftest.py
import pytest
from sqlmodel import SQLModel
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession

@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"

@pytest.fixture
async def async_session():
    engine = create_async_engine("sqlite+aiosqlite://", echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    AsyncSessionLocal = sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )

    async with AsyncSessionLocal() as session:
        yield session

    await engine.dispose()
この構成により、
テストごとに独立したDB
イベントループの競合回避
テスト終了後の確実なリソース解放
が保証されます。

 🔁 FastAPI依存関係のオーバーライド（Async版）FastAPIの依存関係オーバーライドを使い、テスト用AsyncSessionを注入します。
tests/conftest.py
from fastapi.testclient import TestClient
from app.main import app
from app.database import get_session

@pytest.fixture
def client(async_session):
    async def override_get_session():
        yield async_session

    app.dependency_overrides[get_session] = override_get_session
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()
この方法により、本番DBに一切触れず、非同期APIを同期的にテストできます。

 🧪 非同期CRUDテスト最後に、非同期CRUD APIに対するテスト例を示します。
tests/test_items_async.py

def test_create_item(client):
    response = client.post(
        "/items/",
        json={"name": "Async Apple", "price": 200}
    )
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Async Apple"
    assert data["price"] == 200
テストではDBの内部状態を直接検証するのではなく、APIの入出力に集中することが、保守性の高い設計につながります。

 ⚠️ AsyncSessionテストでよくある罠
pytest.mark.asyncio と TestClient を混在させない
セッションをテスト間で使い回しすぎない

expire_on_commit=False を設定し忘れない
engine を fixture の外で共有しない
CI環境でイベントループ設定が異なる点に注意する
これらは、非同期DBテストで頻発するトラブルの原因です。

 ✅ まとめ本記事では、AsyncSessionを使ったFastAPIアプリを 安全・高速・再現性高くテストするための実践構成 を解説しました。
AsyncSession + SQLModel の正しい初期化方法
非同期DBを fixture で完全に分離する設計
FastAPI依存関係の Async オーバーライド
実務で壊れにくい非同期CRUDテストの書き方

 株式会社ONE WEDGE【Serverlessで世の中をもっと楽しく】 ONE WEDGEはServerlessシステム開発を中核技術としてWeb系システム開発、AWS/GCPを利用した業務システム・サービス開発、PWAを用いたモバイル開発、Alexaスキル開発など、元気と技術力を武器にお客様に真摯に向き合う価値創造企業です。

https://onewedge.co.jp/
🌐 はじめに

🧩 使用技術と前提

🗂️ ディレクトリ構成

🛢️ SQLModel + AsyncSession のDB設定

🧪 AsyncSession対応のモデル定義

🔄 非同期CRUDエンドポイント

🧪 テスト用Async DBをfixtureで構築

🔁 FastAPI依存関係のオーバーライド（Async版）

🧪 非同期CRUDテスト

⚠️ AsyncSessionテストでよくある罠

✅ まとめ

株式会社ONE WEDGE

Discussion
