FastAPIでDB操作するユニットテストを書いてみた

2021年12月2日 00:13

この記事は、NAVITIME JAPAN Advent Calendar 2021の 2日目の記事です。

https://adventar.org/calendars/6340

こんにちは。なかぱーです。ナビタイムジャパンで地点検索データの導入や運用基盤の開発を担当しています。

今日は、地点データを管理するWEBアプリケーションのリプレースに関わるなかで気づいたこと、実際に導入してみたことについてを皆さんにご紹介させていただきます。

はじめに

ナビタイムジャパンでは日本全国の観光地やお店、飲食店など多数の地点検索データを扱っており、ユーザーの皆さんが快適な移動を実現できるようデータを収集、管理しています。
これらのデータは様々なお店や企業の方と連携して収集していますが、日々修正や追加が行われており、常に鮮度の高い情報をお届けできるようになっています。
今回、それらのデータ修正や更新をするためのWEBアプリケーションをリプレースすることとなりました。リプレースにあたり、バックエンドフレームワークについて今まで使っていたDjangoではなくFastAPIを採用することになりました。

FastAPIとは

python製のWEBフレームワークで、シンプルで直感的なAPI開発ができるライブラリです。便利なところはたくさんあるのですが、

・pydanticを利用した型安全な開発ができる
・レスポンスのスキーマ定義に合わせて自動的にSwaggerでドキュメントが作られる
・公式ドキュメントがしっかり整理されていてAPI開発初心者でも手をつけやすい

というところに着目し、開発に採用することとなりました。
FastAPIについてはこちらの記事でも紹介しているので、ぜひご覧ください。

リプレースするにあたって困ったこと

上に書いた通り、FastAPIは型を使った開発をサポートしてくれる機能が充実しており、またドキュメントを自動で生成してくれるのでフロント開発者とのコミュニケーションがスムーズにできるなど助けられることはたくさんあるのですが、一方でリプレースする際に既存のツールの細かい仕様やビジネスロジックを考慮できておらず、結果切り戻しや修正に時間を取られることが多くありました。

また、それらが適切に動くことをどうやって保証するのか、何を持って保証したと言えるのかが明確に定まっていませんでした。限られた時間の中でリプレースを終わらせるには、何か打開策を考える必要がありました。

そうだ、テストを書けばいいじゃないか

このような開発の課題を乗り越える方法として、
DB操作の挙動を保証するテストをpytestとDockerを使って実現しました。

具体的にはテストケース用にDockerでモックDBを作成し、本番環境に影響を与えないようなユニットテストを実装しました。これによって、それぞれのビジネスロジックが期待値どおりの処理をすることをコードを通して保証することができます。

このようなテストケースを作成する場合、DockerコンテナにDBを立てる方法に加え、ファイルベースで簡単に利用できるSQLiteをDBとして利用することもあります。今回は、冒頭で述べた通りかなり多くのデータが含まれているテーブルを複数結合して返却するケースのテストを行うことから、Dockerを採用しています。

import pytest
from uuid import uuid4

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker, scoped_session
from sqlalchemy.orm.session import close_all_sessions
from sqlalchemy.exc import SQLAlchemyError

from sample_app.database import Base, get_db
from　　sample_app.main import app

# テスト用のセッションクラスを作成する
class TestingSession(Session):
 def commit(self):
     self.flush()
     self.expire_all()
     
@pytest.fixture(scope="function")
def test_db():
 TEST_DB_URL = "mysql+pymysql://user:pass@127.0.0.1:3306/test_db"
 engine = create_engine(TEST_DB_URL)
 Base.metadata.create_all(bind=engine)

 function_scope = uuid4().hex
 TestingSessionLocal = scoped_session(
     sessionmaker(
         class_=TestingSession,
         autocommit=False,
         autoflush=False,
         bind=engine
     ),
     scopefunc=lambda: function_scope,
 )
 Base.query = TestingSessionLocal
 
 def get_db_for_testing():
   db = TestingSessionLocal()
   try:
       yield db
       db.commit()
   except SQLAlchemyError:
       db.rollback()
　　
　　#　テスト時に依存するDBを本番用からテスト用のものに切り替える 
 app.dependency_overrides[get_db] = get_db_for_testing
 yield TestingSessionLocal()
 
 TestingSessionLocal. remove()　
 close_all_sessions()
 # セッション終了後にengineを破棄し、DBの状態を初期化する
 engine.dispose()

テストコードを作成するにあたってpytestのfixtureという機能を利用しています。fixtureはテスト実行時の前処理や後処理を実行してくれる機能なのですが、これを使うことでテスト用のDBを関数単位でセットアップしてくれるようになります。関数単位でDBを初期化するような記述をすることで、それぞれのテストケースに互いに影響がでないようにすることができます。詳しくはこちらをご参照ください。

テストを書いてみる

それでは実際にAPIの処理をテストしてみましょう。説明にあたり、簡易的なAPIをmain.pyに実装します。なお、説明の便宜上CRUD処理やDBのスキーマ設定などを一つのファイルにまとめていますが、実際の開発の際にはそれぞれ分けて記述しています。

# main.py
from sqlalchemy import Integer, String, Column
from sqlalchemy.orm import Session

from fastapi import Depends, FastAPI, HTTPException

from .database import Base

app = FastAPI()

class User(Base):
   __tablename__ = "users"
   id = Column(Integer, primary_key=True, index=True, autoinclement=True)
   name = Column(String)
   email = Column(String, unique=True)


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


def create_user(db: Session, user: UserCreate):
   db_user = User(email=user.email, name=user.name)
   db.add(db_user)
   db.commit()
   db.refresh(db_user)
   return db_user


@app.post("/users/", response_model=.User)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
   db_user = ['Chikamoto', 'Kinami', 'Itohara']
   if db_user:
       raise HTTPException(status_code=400, detail="User already registered")
   return crud.create_user(db=db, user=user)

create_userをテストするコードをtest.pyに記述します

# test.py
from fastapi.testclient import TestClient

from .main import app
client = TestClient(app)

def test_create_user():
   response = client.post(
       "/users/",
       json={"name": "oyama", "email": "xxxx@xxxx.co.jp"},
   )
   assert response.status_code == 200, response.text
   data = response.json()
   assert data["name"] == "oyama"
   assert "id" in data

pytestを実行すると、テストがうまくいったかどうかを確認することができます。これで実装が終わった後に「あれ、あのビジネスロジックちゃんと考慮できてるかな」「あの仕様ちゃんと守れているかな」と不安になることは無くなりました。

$ pytest -q test.py
...                                                                      [100%]
1 passed in 0.01 seconds

今回ご紹介しているコードはnote用に用意したサンプルコードなのですが、実際にAPIをリリースするにあたりテストケースを調査している際に、開発時には考慮できていなかったビジネスロジックの存在に気づくことができたことで切り戻しを防ぐことができました。エンジニアと利用者のやり取りだけでは防ぐことのできない仕様の抜け漏れは、テストで防ぐことができるんだと痛感しました。

おわりに

今回はPythonのFastAPIを使ったAPIを題材としてお話させていただきましたが、きっと他のフレームワークやアプリケーションにも似たような話があるのではないかなと考えています。

この記事が誰かのお役に立てることがあれば開発者としてこれ以上の幸せはありません。最後まで読んでくださりありがとうございました！