Navigation

📁 Python

FastAPI + SQLAlchemy 2.0 で非同期データベース操作

📅 7月 8, 2025
👤
⏱ 2 min read

FastAPI + SQLAlchemy 2.0 で非同期データベース操作

概要

FastAPIとSQLAlchemy 2.0を組み合わせることで非同期データベース操作を実現する方法について解説します。非同期処理を活用することで、Webアプリケーションのパフォーマンスを向上させることができます。

解決する問題や課題

従来の同期処理では、データベース操作などのI/O処理が待ち時間を引き起こし、アプリケーション全体のレスポンスタイムが遅くなる可能性があります。非同期処理を導入することで、これらの待ち時間を最小限に抑え、ユーザーエクスペリエンスを向上させることができます。

前提知識・必要ライブラリ

  • Pythonの基本知識
  • FastAPIの基本知識
  • SQLAlchemyの基本知識
  • Python 3.7以上
  • FastAPI
  • SQLAlchemy 2.0

環境構築


# 必要ライブラリのインストール
pip install fastapi sqlalchemy

実装コード


from fastapi import FastAPI
from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker

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

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)

async def create_user(db: AsyncSession, name: str):
    async with db() as session:
        user = User(name=name)
        session.add(user)
        await session.commit()

async def get_users(db: AsyncSession):
    async with db() as session:
        return await session.execute(select(User)).scalars().all()

app = FastAPI()

@app.on_event("startup")
async def startup():
    engine = create_async_engine(DATABASE_URL, echo=True)
    async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)

@app.get("/")
async def read_root():
    async with AsyncSession() as session:
        users = await get_users(session)
        return {"users": users}

@app.post("/user/{name}")
async def create_user_endpoint(name: str):
    async with AsyncSession() as session:
        await create_user(session, name)
        return {"message": "User created successfully"}

使用例


# FastAPIアプリケーションを起動して動作を確認
# http://127.0.0.1:8000/ にアクセスしてユーザー一覧を取得し、新しいユーザーを作成する

テストコード


import pytest

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncSession() as session:
        await create_user(session, "Alice")
        users = await get_users(session)
        assert len(users) == 1
        assert users[0].name == "Alice"

応用・カスタマイズ

FastAPIとSQLAlchemyを組み合わせることで、データベース操作を非同期に行うことができますが、パフォーマンスを最適化するためには以下の点に注意すると良いでしょう。

  • クエリの最適化
  • インデックスの活用
  • コネクションプーリングの設定

関連技術

  • FastAPI: https://fastapi.tiangolo.com/
  • SQLAlchemy: https://www.sqlalchemy.org/
  • SQLAlchemy Core 2.0 Documentation: https://docs.sqlalchemy.org/en/20/

—

この記事では、FastAPIとSQLAlchemyを使用して非同期データベース操作を行う方法を紹介しました。非同期処理を活用することで、Webアプリケーションのパフォーマンス向上に貢献することができます。

← Back to Python