FastAPI SQLAlchemy async – 非同期データベース接続を実装する

2026年9月20日

FastAPI SQLAlchemy async – 非同期データベース接続を実装する

同期 DB と非同期 DB の違い

FastAPI は「非同期フレームワーク」ですが、多くの初心者が DB アクセスで同期コードを使ってしまい、非同期のメリットを活かせていません。

今回は、FastAPI と SQLAlchemy async を組み合わせて、「真の非同期 DB アクセス」を実装する方法についてまとめておきたいと思います。

もしご自身で先に公式ドキュメントをご覧になられる方は下のリンクから眺めてみてください。
Async I/O – SQLAlchemy ー公式ドキュメント（英語）

対象となる方

FastAPI を使っている方
DB アクセスの性能を改善したい方
非同期プログラミングを深く学びたい方

　※ このドキュメントは FastAPI 0.100 以上、SQLAlchemy 2.0 以上で書いていきます。

同期 DB vs 非同期 DB の比較

同期コード（ブロッキング）

from sqlalchemy import create_engine

engine = create_engine("postgresql://user:pass@localhost/db")

@app.get("/api/users")
async def get_users():
    with Session(engine) as session:
        # ← この行でブロック
        # DB からのレスポンスを待つまで、他のリクエストは処理できない
        users = session.query(User).all()
    return {"users": users}

非同期コード（ノンブロッキング）

from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")

@app.get("/api/users")
async def get_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    # ← await で「一時停止」、他のリクエストを処理
    users = result.scalars().all()
    return {"users": users}

SQLAlchemy async のセットアップ

SQLAlchemy async を使うには、以下をインストールします。

pip install sqlalchemy asyncpg

その後、DB 接続を設定します。

from sqlalchemy.ext.asyncio import (
    create_async_engine,
    async_sessionmaker,
    AsyncSession,
)
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/db",
    echo=False,
    poolclass=NullPool,
)

async_session_maker = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

async def get_db() -> AsyncSession:
    async with async_session_maker() as session:
        yield session

実装例：CRUD 操作を非同期で

では、ユーザー CRUD の例を見てみましょう。

from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from database import get_db
from models import User
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    email: str
    name: str

@app.get("/api/users")
async def list_users(db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User))
    users = result.scalars().all()
    return {"users": users}

@app.post("/api/users")
async def create_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    new_user = User(email=user_data.email, name=user_data.name)
    db.add(new_user)
    await db.commit()
    await db.refresh(new_user)
    return {"id": new_user.id, "email": new_user.email}

Lambda 環境でのベストプラクティス

Lambda で FastAPI を動かす場合、DB 接続管理に注意が必要です。

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.pool import NullPool

engine = create_async_engine(
    "postgresql+asyncpg://...",
    poolclass=NullPool,  # Lambda 向けの設定
    connect_args={
        "prepared_statement_name_func": lambda: f"__stmt_{uuid4()}__",
        "statement_cache_size": 0,
    },
)

# NullPool を使う理由：
# - Lambda のコンテナは一時的で、再利用時に DB 接続が閉じられている
# - コネクションプールを持つと、スタレな接続を握り続ける
# - NullPool で毎回新しい接続を作り直すのが安全