FastAPIの自動テスト入門：pytestとTestClient、テストDBを一貫して動かす

FastAPIの自動テストは、pytestでテストを集め、TestClientでAPIへリクエストを送り、依存関係をテスト用データベースへ差し替えるところから始めます。

この3点を同じフィクスチャで結ぶと、本番データへ触れず、テストごとに同じ条件を再現できます。

この記事ではTodo ​APIを題材に、最小構成から正常系、失敗系、非同期処理、カバレッジまでを順番に組み立てます。

この記事で作るテスト基盤

• pytest：テストの検出、実行、結果表示を担うPython向けテストフレームワークです。
• TestClient：FastAPIアプリへ同期的にHTTPリクエストを送り、レスポンスを検証するクライアントです。
• フィクスチャ：テスト前の準備と終了後の片付けを再利用するpytestの仕組みです。
• 依存関係の差し替え：本番用のDB接続や認証処理を、テスト専用の実装へ置き換える方法です。

最初に作るべきものは、機能を網羅する大量のテストではありません。

APIを起動し、データを1件作成し、取得結果を確かめられる短い経路です。

テストの種類と役割

テストの種類を分けると、失敗したときに調べる範囲が狭くなります。

TestClientでアプリを直接呼ぶテストは、Webサーバーやリバースプロキシを経由しません。

したがって、アプリ内部のAPIテストには向きますが、配置後のネットワーク構成まで保証するものではありません。

準備するファイルとライブラリ

サンプルはPython ​3.10以降、FastAPI、Pydantic ​v2、SQLAlchemy ​2系の書式で統一します。

fastapi-testing/
├─ ​app/
│ ​ ​├─ ​__init__.py
│ ​ ​├─ ​db.py
│ ​ ​├─ ​models.py
│ ​ ​├─ ​schemas.py
│ ​ ​└─ ​main.py
└─ ​tests/
 ​ ​ ​├─ ​conftest.py
 ​ ​ ​├─ ​test_todos.py
 ​ ​ ​└─ ​test_async.py

仮想環境を有効にしたうえで、次の依存ライブラリを入れます。

python ​-m ​pip ​install ​fastapi ​uvicorn ​sqlalchemy ​pydantic ​pytest ​httpx ​anyio ​pytest-cov

既存プロジェクトでは、先にバージョン固定ファイルと互換性を確認してください。

Todo ​APIの最小実装

テストを理解するために必要な範囲だけ、同期型のTodo ​APIを用意します。

入力検証を詳しく確認したい場合は、FastAPIの型ヒントとPydanticによる入力検証も参照できます。

データベース接続

# ​app/db.py
from ​collections.abc ​import ​Generator

from ​sqlalchemy ​import ​create_engine
from ​sqlalchemy.orm ​import ​Session, ​sessionmaker

DATABASE_URL ​= ​'sqlite:///./app.db'

engine ​= ​create_engine(
 ​ ​ ​ ​DATABASE_URL,
 ​ ​ ​ ​connect_args={'check_same_thread': ​False},
)
SessionLocal ​= ​sessionmaker(
 ​ ​ ​ ​bind=engine,
 ​ ​ ​ ​autoflush=False,
 ​ ​ ​ ​expire_on_commit=False,
)

def ​get_db() ​-> ​Generator[Session, ​None, ​None]:
 ​ ​ ​ ​with ​SessionLocal() ​as ​session:
 ​ ​ ​ ​ ​ ​ ​ ​yield ​session

テーブルモデル

# ​app/models.py
from ​datetime ​import ​datetime

from ​sqlalchemy ​import ​Boolean, ​DateTime, ​String, ​func
from ​sqlalchemy.orm ​import ​DeclarativeBase, ​Mapped, ​mapped_column

class ​Base(DeclarativeBase):
 ​ ​ ​ ​pass

class ​Todo(Base):
 ​ ​ ​ ​__tablename__ ​= ​'todos'

 ​ ​ ​ ​id: ​Mapped[int] ​= ​mapped_column(primary_key=True)
 ​ ​ ​ ​title: ​Mapped[str] ​= ​mapped_column(String(200), ​nullable=False)
 ​ ​ ​ ​is_done: ​Mapped[bool] ​= ​mapped_column(
 ​ ​ ​ ​ ​ ​ ​ ​Boolean, ​default=False, ​nullable=False
 ​ ​ ​ ​)
 ​ ​ ​ ​created_at: ​Mapped[datetime] ​= ​mapped_column(
 ​ ​ ​ ​ ​ ​ ​ ​DateTime(timezone=True),
 ​ ​ ​ ​ ​ ​ ​ ​server_default=func.now(),
 ​ ​ ​ ​ ​ ​ ​ ​nullable=False,
 ​ ​ ​ ​)

入出力スキーマ

# ​app/schemas.py
from ​datetime ​import ​datetime

from ​pydantic ​import ​BaseModel, ​ConfigDict, ​Field

class ​TodoCreate(BaseModel):
 ​ ​ ​ ​title: ​str ​= ​Field(min_length=1, ​max_length=200)
 ​ ​ ​ ​is_done: ​bool ​= ​False

class ​TodoRead(TodoCreate):
 ​ ​ ​ ​model_config ​= ​ConfigDict(from_attributes=True)

 ​ ​ ​ ​id: ​int
 ​ ​ ​ ​created_at: ​datetime

String(200)はDB列の定義です。

「タイトルは200文字以内」という入力ルールは、Field(max_length=200)にも明示します。

これにより、長すぎる入力をDBへ渡す前に検証できます。

APIエンドポイント

# ​app/main.py
from ​typing ​import ​Annotated

from ​fastapi ​import ​Depends, ​FastAPI, ​HTTPException, ​Response, ​status
from ​sqlalchemy ​import ​select
from ​sqlalchemy.orm ​import ​Session

from ​app.db ​import ​get_db
from ​app.models ​import ​Todo ​as ​TodoModel
from ​app.schemas ​import ​TodoCreate, ​TodoRead

app ​= ​FastAPI(title='Todos ​API')
DbSession ​= ​Annotated[Session, ​Depends(get_db)]
@app.post('/todos', ​response_model=TodoRead, ​status_code=201)
def ​create_todo(payload: ​TodoCreate, ​db: ​DbSession):
 ​ ​ ​ ​todo ​= ​TodoModel(**payload.model_dump())
 ​ ​ ​ ​db.add(todo)
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​db.refresh(todo)
 ​ ​ ​ ​return ​todo

@app.get('/todos', ​response_model=list[TodoRead])
def ​list_todos(db: ​DbSession):
 ​ ​ ​ ​statement ​= ​select(TodoModel).order_by(TodoModel.id)
 ​ ​ ​ ​return ​db.scalars(statement).all()

@app.get('/todos/{todo_id}', ​response_model=TodoRead)
def ​get_todo(todo_id: ​int, ​db: ​DbSession):
 ​ ​ ​ ​todo ​= ​db.get(TodoModel, ​todo_id)
 ​ ​ ​ ​if ​todo ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail='Todo ​not ​found')
 ​ ​ ​ ​return ​todo
@app.put('/todos/{todo_id}', ​response_model=TodoRead)
def ​update_todo(todo_id: ​int, ​payload: ​TodoCreate, ​db: ​DbSession):
 ​ ​ ​ ​todo ​= ​db.get(TodoModel, ​todo_id)
 ​ ​ ​ ​if ​todo ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail='Todo ​not ​found')
 ​ ​ ​ ​for ​name, ​value ​in ​payload.model_dump().items():
 ​ ​ ​ ​ ​ ​ ​ ​setattr(todo, ​name, ​value)
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​db.refresh(todo)
 ​ ​ ​ ​return ​todo

@app.delete('/todos/{todo_id}', ​status_code=204)
def ​delete_todo(todo_id: ​int, ​db: ​DbSession) ​-> ​Response:
 ​ ​ ​ ​todo ​= ​db.get(TodoModel, ​todo_id)
 ​ ​ ​ ​if ​todo ​is ​None:
 ​ ​ ​ ​ ​ ​ ​ ​raise ​HTTPException(status_code=404, ​detail='Todo ​not ​found')
 ​ ​ ​ ​db.delete(todo)
 ​ ​ ​ ​db.commit()
 ​ ​ ​ ​return ​Response(status_code=status.HTTP_204_NO_CONTENT)

この例では、アプリ起動時のテーブル作成処理を省いています。

ローカルでアプリを動かす前にマイグレーションを適用し、テストでは後述のフィクスチャで専用テーブルを作成します。

テストDBとTestClientを同じフィクスチャで結ぶ

メモリ上のSQLiteは、接続が変わると別のデータベースとして扱われる場合があります。

そこで、この直列実行向けサンプルではStaticPoolを使い、1テスト内のセッションとTestClientが同じ接続を共有するようにします。

# ​tests/conftest.py
from ​collections.abc ​import ​Iterator

import ​pytest
from ​fastapi.testclient ​import ​TestClient
from ​sqlalchemy ​import ​create_engine
from ​sqlalchemy.orm ​import ​Session
from ​sqlalchemy.pool ​import ​StaticPool

from ​app.db ​import ​get_db
from ​app.main ​import ​app
from ​app.models ​import ​Base

@pytest.fixture
def ​db_session() ​-> ​Iterator[Session]:
 ​ ​ ​ ​test_engine ​= ​create_engine(
 ​ ​ ​ ​ ​ ​ ​ ​'sqlite://',
 ​ ​ ​ ​ ​ ​ ​ ​connect_args={'check_same_thread': ​False},
 ​ ​ ​ ​ ​ ​ ​ ​poolclass=StaticPool,
 ​ ​ ​ ​)
 ​ ​ ​ ​Base.metadata.create_all(test_engine)

 ​ ​ ​ ​with ​Session(test_engine) ​as ​session:
 ​ ​ ​ ​ ​ ​ ​ ​yield ​session

 ​ ​ ​ ​Base.metadata.drop_all(test_engine)
 ​ ​ ​ ​test_engine.dispose()

@pytest.fixture
def ​client(db_session: ​Session) ​-> ​Iterator[TestClient]:
 ​ ​ ​ ​def ​override_get_db():
 ​ ​ ​ ​ ​ ​ ​ ​yield ​db_session

 ​ ​ ​ ​app.dependency_overrides[get_db] ​= ​override_get_db
 ​ ​ ​ ​try:
 ​ ​ ​ ​ ​ ​ ​ ​with ​TestClient(app) ​as ​test_client:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​yield ​test_client
 ​ ​ ​ ​finally:
 ​ ​ ​ ​ ​ ​ ​ ​app.dependency_overrides.clear()

db_sessionはテストごとに新しいDBを作るため、前のテストのデータが残りません。

clientは同じセッションをFastAPIへ渡し、終了時に依存関係の差し替えを必ず解除します。

アプリが起動時と終了時の処理を持つ場合に備え、TestClientはwith文で開いています。

最初のAPIテスト

作成と取得を一つのテストにすると、リクエスト、DB保存、レスポンス変換のつながりを短く確認できます。

# ​tests/test_todos.py
def ​test_create_and_get_todo(client):
 ​ ​ ​ ​created ​= ​client.post(
 ​ ​ ​ ​ ​ ​ ​ ​'/todos',
 ​ ​ ​ ​ ​ ​ ​ ​json={'title': ​'書類を出す'},
 ​ ​ ​ ​)
 ​ ​ ​ ​assert ​created.status_code ​== ​201

 ​ ​ ​ ​created_data ​= ​created.json()
 ​ ​ ​ ​assert ​created_data['title'] ​== ​'書類を出す'
 ​ ​ ​ ​assert ​created_data['is_done'] ​is ​False

 ​ ​ ​ ​fetched ​= ​client.get(f"/todos/{created_data['id']}")
 ​ ​ ​ ​assert ​fetched.status_code ​== ​200
 ​ ​ ​ ​assert ​fetched.json()['id'] ​== ​created_data['id']

このテストが失敗したときは、まずステータスコードを確認し、その後にJSONの各項目を追います。

一つの巨大な比較だけにすると、どの仕様が崩れたのか見分けにくくなります。

フィクスチャで投入したデータをAPIから読む

DBへ準備したデータとAPIが同じセッションを使っているかも検証します。

from ​app.models ​import ​Todo

def ​test_list_todos(db_session, ​client):
 ​ ​ ​ ​db_session.add_all([
 ​ ​ ​ ​ ​ ​ ​ ​Todo(title='A'),
 ​ ​ ​ ​ ​ ​ ​ ​Todo(title='B', ​is_done=True),
 ​ ​ ​ ​])
 ​ ​ ​ ​db_session.commit()

 ​ ​ ​ ​response ​= ​client.get('/todos')

 ​ ​ ​ ​assert ​response.status_code ​== ​200
 ​ ​ ​ ​assert ​[item['title'] ​for ​item ​in ​response.json()] ​== ​['A', ​'B']

元のDBへデータを入れながら、APIだけ別のDBを読む構成では、このテストは通りません。

データ投入用フィクスチャと依存関係の差し替えを同じdb_sessionへ結ぶことが必要です。

失敗系と境界値を仕様として残す

正常系だけでは、入力が不正なときの契約を確認できません。

存在しないIDと長すぎるタイトルを例に、404と422を明示します。

def ​test_get_todo_not_found(client):
 ​ ​ ​ ​response ​= ​client.get('/todos/9999')

 ​ ​ ​ ​assert ​response.status_code ​== ​404
 ​ ​ ​ ​assert ​response.json() ​== ​{'detail': ​'Todo ​not ​found'}

def ​test_rejects_too_long_title(client):
 ​ ​ ​ ​response ​= ​client.post(
 ​ ​ ​ ​ ​ ​ ​ ​'/todos',
 ​ ​ ​ ​ ​ ​ ​ ​json={'title': ​'a' ​* ​201},
 ​ ​ ​ ​)

 ​ ​ ​ ​assert ​response.status_code ​== ​422

FastAPIのリクエスト検証に失敗した場合、この構成では422が返ります。

「不正入力はすべて400」と決めつけず、実際のAPI契約に合わせて期待値を書きます。

型定義と依存性注入をAPI契約として扱う方法を合わせて読むと、何をテストへ残すべきか整理しやすくなります。

更新と削除を一つの利用経路で確かめる

def ​test_update_and_delete(client):
 ​ ​ ​ ​created ​= ​client.post('/todos', ​json={'title': ​'初期'})
 ​ ​ ​ ​todo_id ​= ​created.json()['id']

 ​ ​ ​ ​updated ​= ​client.put(
 ​ ​ ​ ​ ​ ​ ​ ​f'/todos/{todo_id}',
 ​ ​ ​ ​ ​ ​ ​ ​json={'title': ​'更新後', ​'is_done': ​True},
 ​ ​ ​ ​)
 ​ ​ ​ ​assert ​updated.status_code ​== ​200
 ​ ​ ​ ​assert ​updated.json()['title'] ​== ​'更新後'
 ​ ​ ​ ​assert ​updated.json()['is_done'] ​is ​True

 ​ ​ ​ ​deleted ​= ​client.delete(f'/todos/{todo_id}')
 ​ ​ ​ ​assert ​deleted.status_code ​== ​204
 ​ ​ ​ ​assert ​deleted.content ​== ​b''

 ​ ​ ​ ​fetched ​= ​client.get(f'/todos/{todo_id}')
 ​ ​ ​ ​assert ​fetched.status_code ​== ​404

削除処理では204だけでなく、レスポンス本文が空であることと、その後に取得できないことも確かめます。

AsyncClientで非同期テストを書く

テスト関数内で非同期DBや別の非同期関数を直接待機したい場合は、HTTPXのAsyncClientを使います。

FastAPIアプリを直接渡すのではなく、ASGITransportを介して接続します。

# ​tests/test_async.py
import ​pytest
from ​httpx ​import ​ASGITransport, ​AsyncClient

from ​app.db ​import ​get_db
from ​app.main ​import ​app

@pytest.fixture
def ​anyio_backend():
 ​ ​ ​ ​return ​'asyncio'

@pytest.mark.anyio
async ​def ​test_list_todos_async(db_session):
 ​ ​ ​ ​def ​override_get_db():
 ​ ​ ​ ​ ​ ​ ​ ​yield ​db_session

 ​ ​ ​ ​app.dependency_overrides[get_db] ​= ​override_get_db
 ​ ​ ​ ​try:
 ​ ​ ​ ​ ​ ​ ​ ​transport ​= ​ASGITransport(app=app)
 ​ ​ ​ ​ ​ ​ ​ ​async ​with ​AsyncClient(
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​transport=transport,
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​base_url='http://test',
 ​ ​ ​ ​ ​ ​ ​ ​) ​as ​client:
 ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​ ​response ​= ​await ​client.get('/todos')
 ​ ​ ​ ​finally:
 ​ ​ ​ ​ ​ ​ ​ ​app.dependency_overrides.clear()

 ​ ​ ​ ​assert ​response.status_code ​== ​200
 ​ ​ ​ ​assert ​response.json() ​== ​[]

AsyncClientだけでは、アプリのlifespan処理は自動で実行されません。

起動時にDB接続や共有リソースを準備するアプリでは、その処理をテストでどう開始するかを別途決めます。

設計の背景は、FastAPIのlifespanで共有リソースを管理する方法で確認できます。

認証と外部サービスは境界で差し替える

認証サービス、メール送信、決済、外部APIを毎回呼ぶと、テストが遅くなり、ネットワーク状態にも左右されます。

FastAPIの依存関係として切り出せる処理は、app.dependency_overridesでテスト用実装へ差し替えます。

@pytest.fixture
def ​as_admin():
 ​ ​ ​ ​def ​fake_current_user():
 ​ ​ ​ ​ ​ ​ ​ ​return ​{'username': ​'alice', ​'role': ​'admin'}

 ​ ​ ​ ​app.dependency_overrides[get_current_user] ​= ​fake_current_user
 ​ ​ ​ ​try:
 ​ ​ ​ ​ ​ ​ ​ ​yield
 ​ ​ ​ ​finally:
 ​ ​ ​ ​ ​ ​ ​ ​app.dependency_overrides.pop(get_current_user, ​None)

この例のget_current_userは、アプリ側にある認証依存を表します。

差し替え後は、管理者として成功する経路だけでなく、未認証の401と権限不足の403も別のテストで確かめます。

モックするのは自分の処理すべてではなく、外部との境界です。

そうすれば、テスト対象の判断ロジックは実際に動かしながら、副作用だけを抑えられます。

データを初期化する3つの方法

テストデータの後片付けには複数の方法があり、DBと実行方式に応じて選びます。

入門段階では、この記事のようにテストごとに小さなメモリDBを作る方法が追いやすい構成です。

テスト数が増えて遅くなったら、計測してからトランザクション方式や専用DBへ移します。

テスト名とArrange、Act、Assert

テストは失敗したときに読まれるコードです。

名前だけで前提、操作、期待結果がわかるようにします。

• Arrange：Todoや認証状態などの前提を準備します。
• Act：APIへ1回の主な操作を行います。
• Assert：ステータスコード、本文、DBの状態を検証します。

準備が長くなる場合はフィクスチャやファクトリへ移します。

ただし、テストの意図まで隠れるほど抽象化すると、失敗時に追いにくくなります。

カバレッジとCIの使い方

未実行の行を確認するには、pytest-covを使います。

pytest ​--cov=app ​--cov-report=term-missing ​-q

カバレッジ率は、どの行が通っていないかを探す手がかりです。

高い数値だけでは、境界値や失敗時の振る舞いまで検証できているとは限りません。

まずは認証、権限、データ更新、外部連携など、障害時の影響が大きい経路を優先します。

実行時間を調べるときは、次のコマンドで遅いテストを確認できます。

pytest ​--durations=10 ​-q

ローカルで安定したら、プルリクエストごとに同じコマンドをCIで実行します。

テストファイルが増えた段階では、FastAPIのプロジェクト構成を保守しやすく分ける方法も役立ちます。

よくある失敗と確認箇所

導入を6段階に分ける

1. 起動確認かヘルスチェックを1本書きます。
2. 代表的な作成と取得の経路を追加します。
3. 404、422、401、403などの失敗系を仕様として残します。
4. テスト用DBと依存関係の差し替えをフィクスチャへまとめます。
5. 外部サービスとの境界をモックし、タイムアウトや失敗時の処理を確かめます。
6. カバレッジと実行時間を見ながら、CIで毎回実行します。

最初からすべての経路を埋める必要はありません。

変更する機能の正常系と失敗系を追加し、バグ修正では再現テストを先に残すと、テスト群が実際の仕様に沿って育ちます。

FastAPIの自動テストで押さえる要点

• TestClient、テスト用DB、依存関係の差し替えを同じフィクスチャで結びます。
• 正常系だけでなく、404や422などの失敗系をAPI契約として残します。
• 非同期テストではASGITransportとAsyncClientを使い、lifespanの扱いを別に確認します。
• カバレッジ率より先に、障害時の影響が大きい利用経路を検証します。

次の一歩は、既存APIの代表的なエンドポイントを一つ選び、作成と取得が通る短いテストを書くことです。

その1本が安定して動けば、失敗系、認証、外部連携へ無理なく広げられます。