SQLAlchemyで実現！FastAPIユニットテストにおけるデータベース操作

2024-07-27

FastAPI でユニットテストからデータベースにアクセスする方法

ここでは、SQLAlchemy を使って FastAPI アプリケーションでユニットテストからデータベースにアクセスする方法を 2 通りご紹介します。

テスト用のインメモリ SQLite データベースを使用する

この方法は、最も簡単で軽量な方法です。テストごとに新しいインメモリ SQLite データベースを作成し、テスト終了後に破棄します。

手順:

pytest-postgresql や sqlalchemy-ext-declarative のようなライブラリを使用して、テスト用のインメモリ SQLite データベースを作成します。
FastAPI アプリケーションの app.dependency_overrides ディクショナリを使用して、テスト用のデータベースエンジンを本番用のデータベースエンジンに置き換えます。
テストコードで、テスト用のデータベースエンジンを使用してデータベースにアクセスします。

例:

from sqlalchemy import create_engine
from pytest import fixture

@fixture
def test_db():
    engine = create_engine("sqlite:///:memory:")
    yield engine
    engine.dispose()

@app.dependency_overrides
def override_db(db: Session = Depends(get_db)):
    return test_db()

@app.get("/users")
def get_users(db: Session = Depends(override_db)):
    users = db.query(User).all()
    return users

テスト用の Docker コンテナを使用する

この方法は、本番環境に近い環境でテストを実行したい場合に適しています。テストごとに新しい Docker コンテナを作成し、テスト終了後に破棄します。

docker-compose を使用して、テスト用の PostgreSQL または MySQL データベースコンテナを定義します。

import os
from sqlalchemy import create_engine
from pytest import fixture

@fixture(scope="session")
def docker_compose():
    os.system("docker-compose up -d")
    yield
    os.system("docker-compose down")

@fixture
def test_db():
    test_db_url = os.environ["TEST_DB_URL"]
    engine = create_engine(test_db_url)
    yield engine
    engine.dispose()

@app.dependency_overrides
def override_db(db: Session = Depends(get_db)):
    return test_db()

@app.get("/users")
def get_users(db: Session = Depends(override_db)):
    users = db.query(User).all()
    return users

どちらの方法を選択するかは、プロジェクトの要件によって異なります。

テストコードでは、本番環境とは異なるスキーマを使用して、本番環境のデータベースを汚染しないようにしてください。
テストコードでは、コミットをロールバックするように設定されているトランザクションを使用して、データベースへの変更を分離してください。
pytest-cov や coverage のようなツールを使用して、テストカバレッジを測定してください。

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from pytest import fixture

# データベースモデル
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    email = Column(String(255))

# FastAPI アプリケーション
app = FastAPI()

# テスト用のインメモリ SQLite データベースを作成する
@fixture
def test_db():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)

# テスト用のデータベースエンジンを本番用のデータベースエンジンに置き換える
@app.dependency_overrides
def override_db(db: Session = Depends(get_db)):
    return test_db()

# ユーザーを取得するエンドポイント
@app.get("/users")
def get_users(db: Session = Depends(override_db)):
    users = db.query(User).all()
    return users

import os
from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from pytest import fixture

# データベースモデル
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    email = Column(String(255))

# FastAPI アプリケーション
app = FastAPI()

# テスト用の Docker コンテナを起動する
@fixture(scope="session")
def docker_compose():
    os.system("docker-compose up -d")
    yield
    os.system("docker-compose down")

# テスト用のデータベース URL を取得する
@fixture
def test_db():
    test_db_url = os.environ["TEST_DB_URL"]
    engine = create_engine(test_db_url)
    yield engine
    engine.dispose()

# テスト用のデータベースエンジンを本番用のデータベースエンジンに置き換える
@app.dependency_overrides
def override_db(db: Session = Depends(get_db)):
    return test_db()

# ユーザーを取得するエンドポイント
@app.get("/users")
def get_users(db: Session = Depends(override_db)):
    users = db.query(User).all()
    return users

注:

テスト用の Docker コンテナを使用するには、Docker がインストールされている必要があります。

pytest-mock のようなライブラリを使用して、データベースモジュールのモックオブジェクトを作成することができます。これにより、実際のデータベースにアクセスせずに、テストをシミュレートすることができます。

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from pytest import fixture
from unittest.mock import Mock

# データベースモデル
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    email = Column(String(255))

# FastAPI アプリケーション
app = FastAPI()

# モックのデータベースエンジンを作成する
@fixture
def mock_db():
    mock_engine = Mock(spec=create_engine)
    mock_session = Mock(spec=Session)
    mock_engine.session.return_value = mock_session
    return mock_engine

# テスト用のデータベースエンジンをモックのデータベースエンジンに置き換える
@app.dependency_overrides
def override_db(db: Session = Depends(get_db)):
    return mock_db().session

# ユーザーを取得するエンドポイント
@app.get("/users")
def get_users(db: Session = Depends(override_db)):
    users = db.query(User).all()
    return users

テストデータベースを使用する

本番環境とは別の専用のテストデータベースを使用することができます。これにより、本番環境のデータベースを汚染することなく、テストを実行することができます。

テスト専用のデータベースを作成します。
FastAPI アプリケーションの構成で、テスト用のデータベース接続情報を使用するように設定します。
テストコードで、テスト用のデータベースにアクセスします。

from fastapi import FastAPI
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from pytest import fixture

# データベースモデル
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(255))
    email = Column(String(255))

# FastAPI アプリケーション
app = FastAPI()

# テスト用のデータベース URL
TEST_DB_URL = "postgresql://user:password@localhost/test_db"

# テスト用のデータベースエンジンを作成する
@fixture
def test_db():
    engine = create_engine(TEST_DB_URL)
    yield engine
    engine.dispose()

# テスト用のデータベース接続情報を使用するように設定する
@app.on_event("startup")
def startup_event():
    app.dependency_overrides["db"] = test_db

# ユーザーを取得するエンドポイント
@app.get("/users")
def get_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return users

テストランナーを使用する

tox や pytest-bdd のようなテストランナーを使用して、テストを自動化することができます。これらのツールは、テスト環境の設定とテストの実行を簡素化することができます。

[tox]
envlist =
    py38

[testenv]
commands =
    pytest

unit-testing sqlalchemy fastapi

エンティティキャッシュでデータベースへのアクセスを減らす：SQLAlchemyのエンティティキャッシュ機能

クエリキャッシュSQLAlchemyは、発行されたSQLクエリとその結果を内部的にキャッシュできます。これは、同じクエリが繰り返し実行される場合に、データベースへのアクセスを減らすのに役立ちます。エンティティキャッシュSQLAlchemyは、エンティティオブジェクトとその関連オブジェクトをキャッシュできます。これは、エンティティが頻繁にアクセスされる場合に、データベースへのアクセスを減らすのに役立ちます。...

caching sqlalchemy cherrypy