FastAPI入門：routerでAPIを整理・分割する方法

By /

はい、承知いたしました。FastAPIのルーター（APIRouter）を用いたAPIの整理・分割方法について、約5000語の詳細な解説記事を執筆します。記事の内容を直接ここに表示します。

FastAPI入門：APIRouterで大規模APIを整理・分割する究極ガイド

はじめに

FastAPIは、PythonでAPIを構築するための、非常に高速かつモダンなWebフレームワークです。スターレット（Starlette）とPydanticに基づいて構築されており、その高いパフォーマンス、直感的なプログラミングモデル、そして自動生成されるAPIドキュメント（Swagger UI / ReDoc）により、近年Python界隈で急速に人気を集めています。

FastAPIの魅力は多岐にわたりますが、中でも特筆すべきは、型ヒントを活用した強力なデータ検証と依存性の注入（Dependency Injection）システムです。これにより、コードの品質が向上し、開発効率が大幅に向上します。しかし、どんなフレームワークを使っても、アプリケーションが成長し、エンドポイントの数が増えるにつれて、コードの管理は難しくなっていきます。

なぜAPIを整理・分割する必要があるのか？

APIエンドポイントが数十、数百と増えてくると、単一のファイルに全てのルーティング定義やロジックを記述することは非現実的になります。このような「一枚岩（Monolithic）」なコード構造は、以下のような様々な問題を引き起こします。

  1. 可読性の低下: ファイルが巨大になり、全体の構造を把握するのが困難になります。特定の機能やエンドポイントを探すのが一苦労になります。
  2. 保守性の低下: コードの変更や機能追加が、他の部分に予期せぬ影響を与える可能性が高まります。デバッグも難しくなります。
  3. チーム開発の困難化: 複数の開発者が同じファイルを同時に編集することになり、コンフリクト（競合）が発生しやすくなります。作業の並行化が難しくなります。
  4. 再利用性の低下: 特定の機能に関連するコードがファイル全体に散在するため、他のプロジェクトやアプリケーション内での再利用が困難になります。
  5. スケーラビリティの問題: コードが密結合になり、機能ごとのスケールアウトが難しくなります。

これらの問題を解決し、より管理しやすく、保守しやすい、そしてチーム開発に適したAPIアプリケーションを構築するためには、コードを機能や責務ごとに分割・整理する必要があります。FastAPIでは、この目的のために APIRouter という非常に便利な機能を提供しています。

この記事で学ぶこと

この記事では、FastAPIの APIRouter を徹底的に解説し、以下の点を深く掘り下げていきます。

  • APIRouter の基本的な使い方：ルーターの定義とアプリケーションへの登録
  • ルーターを使ったAPIエンドポイントの整理方法
  • APIRouter の詳細設定：パスプレフィックス、タグ、依存関係の設定
  • 推奨されるプロジェクト構造と、APIRouter を活用した大規模なアプリケーションの構成方法
  • 依存性の注入（Dependency Injection）を APIRouter と組み合わせて活用する方法
  • エラーハンドリングやミドルウェアをルーターレベルで設定する方法
  • 分割されたAPIのテスト方法
  • 大規模開発における APIRouter の高度な活用戦略（バージョン管理など）
  • APIRouter を使用する上でのベストプラクティスと注意点

この記事を読むことで、あなたはFastAPIを使って、小規模なプロトタイプから、チームで開発する大規模で堅牢なAPIアプリケーションまで、あらゆる規模のプロジェクトに適切に対応できるスキルを習得できるでしょう。

それでは、FastAPIの APIRouter の世界に飛び込んでいきましょう。

FastAPIの基本構造と課題

まず、APIRouter を使わない場合のシンプルなFastAPIアプリケーションの構造を見てみましょう。

通常、FastAPIアプリケーションは一つのPythonファイル（例えば main.py）から始まります。

“`python

main.py (APIRouterを使わない例)

from fastapi import FastAPI

app = FastAPI()

@app.get(“/”)
def read_root():
return {“message”: “Hello, World”}

@app.get(“/users/{user_id}”)
def read_user(user_id: int):
# ユーザー取得ロジック
return {“user_id”: user_id, “name”: f”user_{user_id}”}

@app.post(“/users/”)
def create_user(user: dict):
# ユーザー作成ロジック
print(f”ユーザー作成リクエスト: {user}”)
return {“message”: “User created successfully”, “user”: user}

@app.get(“/items/{item_id}”)
def read_item(item_id: int, q: str | None = None):
# 商品取得ロジック
return {“item_id”: item_id, “q”: q}

@app.put(“/items/{item_id}”)
def update_item(item_id: int, item: dict):
# 商品更新ロジック
print(f”商品更新リクエスト (ID: {item_id}): {item}”)
return {“message”: f”Item {item_id} updated successfully”, “item”: item}

… 他の多くのエンドポイントが続く …

“`

このコードは、ルートパス (/)、ユーザー関連のエンドポイント (/users/...)、商品関連のエンドポイント (/items/...) を定義しています。

この例はまだシンプルですが、もしユーザー管理、商品管理、注文管理、認証、決済処理など、様々な機能に関連するエンドポイントがそれぞれ数十個ずつ追加されたらどうなるでしょうか？ main.py ファイルは数百行、数千行規模に肥大化し、以下の問題が顕在化します。

  • 見通しの悪さ: ユーザー関連のコード、商品関連のコード、その他のコードが混在し、どこに何があるのか分かりづらくなります。
  • 重複コード: 例えば、特定の認証ロジックを複数のエンドポイントで呼び出す場合、それを各エンドポイントの関数内で行うと重複が発生しやすくなります。
  • 密結合: 異なる機能のエンドポイントが同じファイルにあるため、変更が他の機能に影響を与えやすくなります。
  • 保守の困難さ: バグ修正や機能追加の際に、巨大なファイルの中から関連する部分を探し出し、慎重に変更する必要があります。

このような課題を解決するために、FastAPIは APIRouter を提供しています。APIRouter を使うことで、APIエンドポイントを機能別やリソース別に論理的に分割し、それぞれのグループを独立したファイルやモジュールとして管理できるようになります。

APIRouterの基本

APIRouter は、FastAPIアプリケーションのサブアプリケーションのようなものです。これ自体が独立したルーティングテーブルを持ち、そこにエンドポイント（パスオペレーション）を定義することができます。そして、定義したルーターをメインのFastAPIアプリケーションに「マウント」または「インクルード」することで、それらのエンドポイントがアプリケーション全体のルーティングテーブルに追加されます。

APIRouterの基本的な使い方

基本的な使い方の手順は以下の通りです。

  1. APIRouter インスタンスの作成: 各機能（例: ユーザー、商品）に対応するファイルで、fastapi.APIRouter のインスタンスを作成します。
  2. ルーターへのパス定義: 作成したルーターインスタンスに対して、@router.get(), @router.post() などのデコレーターを使ってエンドポイント（パスオペレーション）を定義します。これは FastAPI インスタンスに対して行うのと全く同じ方法です。
  3. FastAPIアプリケーションへのルーター登録: メインの FastAPI アプリケーションインスタンスで、app.include_router() メソッドを使って、定義したルーターを登録します。

具体的なコード例を見てみましょう。ユーザー関連のAPIを users.py に、商品関連のAPIを items.py に分割し、main.py でそれらを統合する例です。

まず、ユーザー関連のAPIを定義するファイル (users.py) を作成します。

“`python

app/routers/users.py

from fastapi import APIRouter, HTTPException
from typing import List, Dict

APIRouterのインスタンスを作成

prefix=”/users” を指定することで、このルーター内で定義される全てのパスに “/users” が付加される

tags=[“users”] を指定することで、このルーター内のエンドポイントがドキュメント上で “users” タグでグループ化される

router = APIRouter(
prefix=”/users”,
tags=[“users”],
responses={404: {“description”: “User not found”}},
)

ダミーデータ (本来はDBなどから取得)

fake_users_db = {
“1”: {“username”: “john_doe”, “full_name”: “John Doe”},
“2”: {“username”: “jane_smith”, “full_name”: “Jane Smith”},
}

@router.get(“/”)
def read_users():
“””
全てのユーザー情報を取得します。
“””
# /users にGETリクエストがあった場合、この関数が実行される
# (prefix=”/users” がここで効いている)
return list(fake_users_db.values())

@router.get(“/{user_id}”)
def read_user(user_id: str):
“””
指定されたIDのユーザー情報を取得します。
“””
# /users/{user_id} にGETリクエストがあった場合、この関数が実行される
if user_id not in fake_users_db:
raise HTTPException(status_code=404, detail=”User not found”)
return fake_users_db[user_id]

@router.post(“/”)
def create_user(user: Dict[str, str]):
“””
新しいユーザーを作成します。
“””
# /users にPOSTリクエストがあった場合、この関数が実行される
# 本来はDBに保存するなどの処理を行う
user_id = str(len(fake_users_db) + 1)
fake_users_db[user_id] = user
return {“message”: “User created successfully”, “user_id”: user_id, “user”: user}

@router.put(“/{user_id}”)
def update_user(user_id: str, user: Dict[str, str]):
“””
指定されたIDのユーザー情報を更新します。
“””
# /users/{user_id} にPUTリクエストがあった場合、この関数が実行される
if user_id not in fake_users_db:
raise HTTPException(status_code=404, detail=”User not found”)
fake_users_db[user_id].update(user)
return {“message”: f”User {user_id} updated successfully”, “user”: fake_users_db[user_id]}

@router.delete(“/{user_id}”)
def delete_user(user_id: str):
“””
指定されたIDのユーザーを削除します。
“””
# /users/{user_id} にDELETEリクエストがあった場合、この関数が実行される
if user_id not in fake_users_db:
raise HTTPException(status_code=404, detail=”User not found”)
del fake_users_db[user_id]
return {“message”: f”User {user_id} deleted successfully”}

“`

次に、商品関連のAPIを定義するファイル (items.py) を作成します。

“`python

app/routers/items.py

from fastapi import APIRouter, HTTPException
from typing import Dict, Optional

router = APIRouter(
prefix=”/items”,
tags=[“items”],
responses={404: {“description”: “Item not found”}},
)

ダミーデータ

fake_items_db = {
“foo”: {“name”: “Foo”, “price”: 50.2},
“bar”: {“name”: “Bar”, “price”: 60.0, “description”: “A very nice Bar”},
“baz”: {“name”: “Baz”, “price”: 40.5, “description”: “The great Baz”},
}

@router.get(“/”)
def read_items(q: Optional[str] = None):
“””
全ての商品情報を取得します。オプションでクエリパラメータ q でフィルタリングできます。
“””
# /items にGETリクエストがあった場合、この関数が実行される
if q:
return {k: v for k, v in fake_items_db.items() if q.lower() in v[“name”].lower()}
return fake_items_db

@router.get(“/{item_id}”)
def read_item(item_id: str):
“””
指定されたIDの商品情報を取得します。
“””
# /items/{item_id} にGETリクエストがあった場合、この関数が実行される
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
return fake_items_db[item_id]

@router.post(“/”)
def create_item(item: Dict):
“””
新しい商品を作成します。
“””
# /items にPOSTリクエストがあった場合、この関数が実行される
item_id = item.get(“name”, “new_item”).lower().replace(” “, “_”)
fake_items_db[item_id] = item
return {“message”: “Item created successfully”, “item_id”: item_id, “item”: item}
“`

最後に、これらのルーターをメインのFastAPIアプリケーションに登録する main.py を作成します。

“`python

app/main.py

from fastapi import FastAPI

他のファイルで定義したルーターをインポート

from .routers import users, items

app = FastAPI(
title=”My Awesome API”,
description=”This is a sample API using FastAPI with APIRouter.”,
version=”0.1.0″,
)

ルーターをアプリケーションに登録

include_router() メソッドを使って、定義したルーターをメインのFastAPIアプリケーションに追加します。

ここで prefix を指定することも可能ですが、通常はルーター定義側で指定します。

app.include_router(users.router)
app.include_router(items.router)

アプリケーション全体のルートパス

@app.get(“/”)
def read_root():
“””
APIのルートエンドポイントです。
“””
return {“message”: “Welcome to My Awesome API!”}

アプリケーションの実行方法 (例: uvicorn app.main:app –reload)

“`

この構成でアプリケーションを起動すると、以下のエンドポイントが利用可能になります。

  • / (main.py で定義)
  • /users/ (users.py で定義)
  • /users/{user_id} (users.py で定義)
  • /items/ (items.py で定義)
  • /items/{item_id} (items.py で定義)

これらのエンドポイントは、FastAPIによって自動生成されるAPIドキュメント（/docs や /redoc）にも反映されます。users.py と items.py で tags を指定したため、ドキュメント上でも「users」と「items」というグループに分かれて表示され、非常に見やすくなります。

このように、APIRouter を使うことで、異なる機能に関連するエンドポイントを物理的に別のファイルに分割し、コードのモジュール化を実現できます。これにより、各ファイルが担当する責務が明確になり、可読性、保守性、チーム開発の効率が向上します。

APIRouterの詳細設定

APIRouter のインスタンスを作成する際には、様々なパラメータを指定してルーターの振る舞いやドキュメント表示をカスタマイズできます。先ほどの例でもいくつか使用しましたが、ここではさらに詳細を見ていきましょう。

python
router = APIRouter(
prefix="/myprefix",
tags=["mytag"],
dependencies=[Depends(my_dependency)],
responses={404: {"description": "Not found"}},
default_response_class=JSONResponse,
redirect_slashes=True,
generate_unique_id_function=my_generate_id_function,
)

主なパラメータは以下の通りです。

prefix: パスプレフィックスの設定

  • prefix="/myprefix" のように指定すると、このルーターで定義される 全ての パスオペレーションの前に指定したパスが付加されます。
  • 先ほどの例では、users.py に prefix="/users" を指定したことで、@router.get("/") は /users/ に、@router.get("/{user_id}") は /users/{user_id} に対応するようになりました。
  • 利点: 各エンドポイントのパス定義が簡潔になります (/ や /{user_id} のように記述できる)。また、関連するエンドポイント群に一貫したパス構造を強制できます。

tags: ドキュメント上のグループ化

  • tags=["mytag"] のように文字列のリストを指定すると、このルーターで定義される 全ての エンドポイントが、自動生成されるAPIドキュメント（Swagger UI / ReDoc）上で指定したタグ名でグループ化されます。
  • 複数のタグを指定することも可能です。
  • 利点: APIドキュメントの可読性が大幅に向上します。ユーザーは機能ごとに関連するエンドポイントを容易に見つけることができます。

dependencies: ルーター全体に共通の依存関係

  • dependencies=[Depends(my_dependency)] のように依存関係のリストを指定すると、このルーターで定義される 全ての エンドポイントに対して、指定した依存関係が自動的に適用されます。
  • 利点: 認証や共通のデータロード、権限チェックなど、複数のエンドポイントで共通して必要な処理を簡単に適用できます。各パスオペレーション関数で Depends() を繰り返し記述する手間が省けます。
  • 注意点: ここで指定した依存関係は、エンドポイントレベルで指定した依存関係よりも 先に 実行されます。

依存関係については後で詳しく掘り下げます。

responses: ルーター全体に共通のレスポンス定義

  • responses={404: {"description": "Not found"}, ...} のように辞書で指定すると、このルーターで定義される 全ての エンドポイントに対して、指定したレスポンス定義が自動的にドキュメントに追加されます。
  • 利点: 共通のエラーレスポンス（例: 認証失敗の401、存在しないリソースの404など）を一度定義すれば、ルーター内の全てのエンドポイントのドキュメントに反映されます。APIの一貫性が向上します。

default_response_class: デフォルトのレスポンスクラス

  • default_response_class=JSONResponse のように指定すると、このルーターで定義されるエンドポイントのデフォルトのレスポンスクラスを変更できます。通常は JSONResponse ですが、XMLResponseなどに変更したい場合に指定します。

redirect_slashes: 末尾スラッシュのリダイレクト

  • redirect_slashes=True (デフォルト) と指定すると、末尾にスラッシュがある/ないパスへのリクエストが、対応するパスに自動的にリダイレクトされます（例: /items へのリクエストが /items/ にリダイレクトされるなど）。
  • redirect_slashes=False とすると、厳密にパスが一致しないと404になります。

generate_unique_id_function: OpenAPIオペレーションID生成関数のカスタマイズ

  • 自動生成されるOpenAPIドキュメントでは、各パスオペレーションにユニークなIDが割り当てられます。デフォルトの生成ロジックをカスタマイズしたい場合に指定します。

これらの詳細設定を適切に使うことで、APIRouter は単なるコード分割のツールにとどまらず、APIの構造設計、ドキュメント生成、共通処理の適用など、様々な面で強力な機能を発揮します。

コード構成とプロジェクト構造

APIRouter を最大限に活用し、保守性の高いFastAPIアプリケーションを構築するためには、適切なプロジェクト構造を採用することが重要です。ここでは一般的な推奨構造を紹介します。

FastAPIアプリケーションの構造は、その規模や複雑さによって異なりますが、一般的なウェブアプリケーションのMVC (Model-View-Controller) や Clean Architecture、DDD (Domain-Driven Design) などの考え方を参考に、責務ごとにコードを分離するのが良いでしょう。

以下に、APIRouter を活用した基本的なプロジェクト構造の例を示します。

.
├── app/
│ ├── __init__.py # パッケージとして認識させるための空ファイル
│ ├── main.py # FastAPIアプリケーションインスタンスの生成、ルーター登録など
│ ├── dependencies.py # 依存性注入で使用する共通関数など
│ ├── models.py # Pydanticモデルの定義 (リクエスト/レスポンスのデータ構造)
│ ├── services/ # ビジネスロジックを記述するサービス層
│ │ ├── __init__.py
│ │ ├── user_service.py
│ │ ├── item_service.py
│ │ └── ...
│ ├── routers/ # APIRouterを使って定義されたエンドポイント群
│ │ ├── __init__.py # ここでルーターをインポートしてリスト化すると便利
│ │ ├── users.py # ユーザー関連ルーター
│ │ ├── items.py # 商品関連ルーター
│ │ └── ...
│ └── database.py # データベース接続設定など (任意)
└── requirements.txt # 依存パッケージリスト
└── .env # 環境変数 (任意)
└── README.md
└── ...

この構造の各ディレクトリ/ファイルの役割は以下の通りです。

  • app/: アプリケーションのルートディレクトリ（Pythonパッケージとして扱われます）。
  • app/__init__.py: app ディレクトリをPythonパッケージとして認識させるための空ファイルです。
  • app/main.py: アプリケーションのエントリーポイントです。FastAPI アプリケーションインスタンスを作成し、include_router() を使って routers/ ディレクトリで定義されたルーターを登録します。アプリケーション全体のミドルウェアやイベントハンドラーなどもここに記述することがあります。
  • app/dependencies.py: 認証情報 (Security や OAuth2PasswordBearer)、データベースセッション、共通の設定値など、依存性注入で使用する関数やクラスを定義します。
  • app/models.py: リクエストボディやレスポンスデータの構造、データベースモデルなどをPydanticやSQLAlchemyなどを使って定義します。
  • app/services/: ビジネスロジックを記述する層です。データベース操作、外部API連携、複雑な計算などをここで行います。ルーターは基本的にこのサービス層を呼び出すだけにすることで、ルーティングとビジネスロジックを分離できます。
  • app/routers/: APIRouter インスタンスを定義し、実際のエンドポイント（パスオペレーション）を記述する層です。各ファイルが特定の機能やリソース（ユーザー、商品など）を担当します。ルーターは services/ 層の関数を呼び出します。
  • app/routers/__init__.py: このファイルで routers ディレクトリ内の各ルーターモジュールをインポートし、リストとしてエクスポートすると、main.py でルーターをまとめて登録する際に便利です。
  • app/database.py: データベースへの接続設定、セッション管理などを記述します。

app/routers/__init__.py を活用したルーター登録

app/routers/__init__.py を以下のように記述することで、main.py でルーターをまとめてインポート・登録できます。

“`python

app/routers/init.py

from fastapi import APIRouter

各ルーターファイルをインポート

from .users import router as users_router
from .items import router as items_router

ここで定義したルーターをリストにまとめる

main.py からこのリストを参照して、まとめて include_router できる

all_routers: list[APIRouter] = [
users_router,
items_router,
# 他のルーターもここに追加
]

このファイル自体は特に何もしないが、パッケージとして機能し、

main.py から app.routers.all_routers としてアクセス可能になる。

(または from .routers import all_routers とインポート)

“`

そして、app/main.py でこのリストを利用します。

“`python

app/main.py (routers/init.py を利用する例)

from fastapi import FastAPI

routers/init.py からルーターのリストをインポート

from .routers import all_routers

app = FastAPI(
title=”My Awesome API”,
description=”This is a sample API using FastAPI with APIRouter.”,
version=”0.1.0″,
)

all_routers リストに含まれる全てのルーターをループで登録

for router in all_routers:
app.include_router(router)

@app.get(“/”)
def read_root():
return {“message”: “Welcome to My Awesome API!”}

… 他の設定 (ミドルウェアなど) …

“`

この方法を使うと、新しいルーターファイルを追加した場合に main.py を変更する必要がなくなり、app/routers/__init__.py にそのルーターを追加するだけで済むようになります。これにより、main.py がシンプルに保たれ、新しい機能の追加が容易になります。

このプロジェクト構造はあくまで一例ですが、このように責任を分離することで、コードのどこに何があるか分かりやすくなり、変更による影響範囲を限定しやすくなります。APIRouter はこの構造の中心的な役割を果たし、ルーティング層を独立したモジュールとして管理できるようにします。

依存性の注入（Dependency Injection）とAPIRouter

FastAPIの最も強力な機能の一つが依存性の注入（Dependency Injection: DI）です。Depends 関数を使うことで、パスオペレーション関数が必要とする「依存関係」（データベースセッション、現在のユーザー、設定値など）を宣言的に定義し、FastAPIがそれらを自動的に解決して関数に渡してくれます。

APIRouter はこのDIシステムと非常にうまく連携します。ルーターレベルで依存関係を定義することで、そのルーターに含まれる全てのエンドポイントに共通の依存関係を簡単に適用できます。

ルーターレベルの依存性

APIRouter のコンストラクタに dependencies パラメータを指定することで、ルーター全体に共通の依存関係を設定できます。

“`python

app/dependencies.py

from fastapi import Header, HTTPException, Depends

ダミーのAPIキー認証依存性

async def verify_api_key(x_api_key: str = Header()):
“””
APIキーを検証する依存性。
“””
if x_api_key != “my_secret_api_key”: # 本番ではセキュアな方法でキーを管理
raise HTTPException(status_code=401, detail=”Invalid API Key”)
# 検証に成功した場合、何も返さない (または必要な値を返す)

ダミーのデータベースセッション依存性 (ここではシンプルな例として扱う)

class DBSession:
def init(self):
print(“データベースセッションを作成しました”)
# 実際のDB接続処理
def close(self):
print(“データベースセッションを閉じました”)
# 実際のDB切断処理

データベースセッションを提供する依存性関数

実際には contextmanager や yield を使ってセッションの開始/終了を制御することが多い

(yieldを使ったDIについてはFastAPI公式ドキュメントを参照)

def get_db_session():
“””
データベースセッションの依存性。
“””
db = DBSession()
try:
yield db # yieldを使うことで、パスオペレーション終了後にfinallyブロックが実行される
finally:
db.close()

app/routers/items.py (dependenciesパラメータを使う例)

from fastapi import APIRouter, Depends, HTTPException
from typing import Dict, Optional

dependencies.py から依存性をインポート

from ..dependencies import verify_api_key, DBSession, get_db_session

このルーターの全てのエンドポイントに対して、verify_api_key と get_db_session が実行される

router = APIRouter(
prefix=”/items”,
tags=[“items”],
dependencies=[Depends(verify_api_key), Depends(get_db_session)], # <— ここで指定
responses={404: {“description”: “Item not found”}},
)

ダミーデータ (実際にはDBから取得)

fake_items_db = {
“foo”: {“name”: “Foo”, “price”: 50.2},
“bar”: {“name”: “Bar”, “price”: 60.0, “description”: “A very nice Bar”},
“baz”: {“name”: “Baz”, “price”: 40.5, “description”: “The great Baz”},
}

このエンドポイントはルーターレベルの依存性 (verify_api_key, get_db_session) と

自身の依存性 (db: DBSession) の両方を持つ

@router.get(“/{item_id}”)
def read_item(item_id: str, db: DBSession = Depends(get_db_session)): # <— エンドポイント固有の依存性
“””
指定されたIDの商品情報を取得します。
ルーターレベルの依存性(APIキー検証, DBセッション)が適用されます。
“””
# ここでは db オブジェクトが利用できる
print(f”DBセッションを使用して商品 {item_id} を取得”)
if item_id not in fake_items_db:
raise HTTPException(status_code=404, detail=”Item not found”)
return fake_items_db[item_id]

ルーターレベルの依存性だけが適用されるエンドポイント

@router.get(“/”)
def read_items(q: Optional[str] = None):
“””
全ての商品情報を取得します。ルーターレベルの依存性が適用されます。
“””
# この関数定義には get_db_session の Depends() はないが、
# ルーターレベルの dependencies で指定されているため実行される
# (ただし、ここでは db オブジェクトはパスオペレーション関数に渡されない)
# パスオペレーション関数内で DB オブジェクトを使う場合は、関数引数にも Depends を指定する必要がある
return fake_items_db
“`

この例では、items ルーターの dependencies に Depends(verify_api_key) と Depends(get_db_session) を指定しています。これにより、/items/ または /items/{item_id} への 全てのリクエスト に対して、リクエストハンドラーが実行される前に verify_api_key 関数と get_db_session 関数が実行されます。

verify_api_key はAPIキーが無効な場合に HTTPException を発生させ、それ以降の処理（パスオペレーション関数や他の依存性）の実行を停止します。get_db_session はデータベースセッションを作成し、yield でセッションオブジェクトを返します。

重要: ルーターレベルの dependencies で yield を含む依存性関数（例: get_db_session）を指定した場合、その依存性は 全てのパスオペレーション関数に対して実行されます。しかし、その依存性によって提供される値（この例では db オブジェクト）をパスオペレーション関数内で利用したい場合は、そのパスオペレーション関数の引数にも明示的に db: DBSession = Depends(get_db_session) のように指定する必要があります。ルーターレベルの dependencies はあくまで「依存性の実行を保証する」ためのものであり、値を関数引数に渡す役割は持ちません。値を渡すのは、関数引数での Depends() の役割です。

DIを活用したロジックの分離（サービス層）

依存性注入は、ビジネスロジックをルーター層から分離し、サービス層に移行させる際に非常に役立ちます。ルーターはリクエストの受け付け、Pydanticモデルによるデータの検証、そしてサービス層の適切な関数を呼び出すことだけに集中します。サービス層は実際のビジネスロジック（データベース操作など）を実行し、結果を返します。

“`python

app/services/user_service.py

from typing import Dict, List

データベースセッションの依存性を利用

from ..dependencies import DBSession

ダミーデータ (ここではDBセッションを使わないが、DIの概念を示すため引数に追加)

fake_users_db = {
“1”: {“username”: “john_doe”, “full_name”: “John Doe”},
“2”: {“username”: “jane_smith”, “full_name”: “Jane Smith”},
}

class UserService:
def init(self, db: DBSession):
# サービスインスタンス生成時にDBセッションを受け取る
self.db = db
print(“UserService インスタンスを作成しました”)

def get_users(self) -> List[Dict]:
    # 実際のDB操作ロジック (例: return self.db.query(User).all())
    print("UserService: 全ユーザーを取得")
    return list(fake_users_db.values())

def get_user_by_id(self, user_id: str) -> Dict | None:
    # 実際のDB操作ロジック (例: return self.db.query(User).filter(User.id == user_id).first())
    print(f"UserService: ユーザー {user_id} を取得")
    return fake_users_db.get(user_id)

def create_user(self, user_data: Dict) -> Dict:
    # 実際のDB操作ロジック (例: new_user = User(**user_data); self.db.add(new_user); self.db.commit())
    print("UserService: ユーザーを作成")
    # ダミーデータへの追加処理 (簡易例)
    user_id = str(len(fake_users_db) + 1)
    fake_users_db[user_id] = user_data
    return {"user_id": user_id, **user_data}

依存性注入でUserServiceインスタンスを提供する関数

def get_user_service(db: DBSession = Depends(get_db_session)):
“””
UserServiceを提供する依存性。
get_db_session 依存性を使ってDBセッションを受け取り、
UserServiceインスタンスを生成して返す。
“””
print(“get_user_service 依存性を実行”)
return UserService(db=db)

app/routers/users.py (サービス層とDIを使う例)

from fastapi import APIRouter, Depends, HTTPException
from typing import List, Dict

サービスと依存性をインポート

from ..services.user_service import UserService, get_user_service
from ..dependencies import verify_api_key # 必要であれば共通依存性を適用

router = APIRouter(
prefix=”/users”,
tags=[“users”],
dependencies=[Depends(verify_api_key)], # APIキー認証はルーターレベルで適用
responses={404: {“description”: “User not found”}},
)

@router.get(“/”)
def read_users(user_service: UserService = Depends(get_user_service)): # <– UserServiceをDI
“””
全てのユーザー情報を取得します。
“””
# パスオペレーション関数はUserServiceインスタンスを受け取り、メソッドを呼び出すだけ
print(“Router: read_users 処理開始”)
users = user_service.get_users()
print(“Router: read_users 処理終了”)
return users

@router.get(“/{user_id}”)
def read_user(user_id: str, user_service: UserService = Depends(get_user_service)): # <– UserServiceをDI
“””
指定されたIDのユーザー情報を取得します。
“””
print(f”Router: read_user (ID: {user_id}) 処理開始”)
user = user_service.get_user_by_id(user_id)
print(f”Router: read_user (ID: {user_id}) 処理終了”)
if user is None:
raise HTTPException(status_code=404, detail=”User not found”)
return user

… 他のエンドポイントも同様に UserService を Depends で取得して利用 …

“`

この構造の利点は以下の通りです。

  • 責務の分離: ルーターはHTTPリクエスト/レスポンスとルーティング、Pydanticモデルによる入出力検証に専念します。ビジネスロジックはサービス層にカプセル化されます。
  • テストの容易性: サービス層はHTTPの依存性から切り離されているため、単体テストが非常に容易になります。ルーターのテストでは、サービス層をモック（Mock）してテストできます。
  • 再利用性: サービス層のロジックは、APIエンドポイントだけでなく、他の部分（例えばバッチ処理など）からも再利用しやすくなります。
  • コードの見通し: 各レイヤー（ルーター、サービス、データベースなど）が独立しているため、コード全体の構造が分かりやすくなります。

APIRouter は、この依存性注入を活用した多層アーキテクチャを構築する上で、ルーティング層を独立させるための基盤を提供します。

エラーハンドリングとAPIRouter

FastAPIでは、例外ハンドラーを定義することで、特定のエラーや例外が発生した際にカスタムのレスポンスを返すことができます。これは RequestValidationException や HTTPException など、FastAPIが提供する例外だけでなく、カスタム例外や標準のPython例外に対しても設定可能です。

例外ハンドラーはアプリケーション全体 (app.add_exception_handler()) または APIRouter レベル (router.add_exception_handler()) で設定できます。

ルーター固有の例外ハンドラー

特定のルーター内で発生した例外に対してのみカスタムハンドリングを適用したい場合に、router.add_exception_handler() を使用します。

“`python

app/routers/items.py (例外ハンドラーの例)

from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import JSONResponse
from typing import Dict, Optional

router = APIRouter(
prefix=”/items”,
tags=[“items”],
responses={404: {“description”: “Item not found”}},
)

ダミーデータ

fake_items_db = {
“foo”: {“name”: “Foo”, “price”: 50.2},
“bar”: {“name”: “Bar”, “price”: 60.0, “description”: “A very nice Bar”},
“baz”: {“name”: “Baz”, “price”: 40.5, “description”: “The great Baz”},
}

カスタム例外を定義 (例)

class ItemCreationError(Exception):
def init(self, item_name: str):
self.item_name = item_name

このルーター固有の例外ハンドラーを定義

@router.exception_handler(ItemCreationError)
async def item_creation_exception_handler(request: Request, exc: ItemCreationError):
“””
ItemCreationError が発生した場合のカスタムハンドラー。
“””
print(f”Caught ItemCreationError for item: {exc.item_name}”)
return JSONResponse(
status_code=400, # Bad Request
content={“message”: f”Failed to create item: {exc.item_name}”},
)

@router.exception_handler(HTTPException)
async def custom_http_exception_handler(request: Request, exc: HTTPException):
“””
このルーター内で発生した HTTPException を補足するハンドラー。
グローバルな HTTPException ハンドラーより優先される。
“””
print(f”Caught HTTPException in items router: {exc.status_code}, Detail: {exc.detail}”)
# ここでカスタムのログ出力や、エラー詳細の整形などを行う
# デフォルトの HTTPException ハンドラーを使う場合は、FastAPIの例外ハンドラーを再利用する
# from fastapi.exception_handlers import http_exception_handler
# return await http_exception_handler(request, exc)

 # または独自のカスタムレスポンスを返す
 return JSONResponse(
     status_code=exc.status_code,
     content={"message": f"An error occurred with item resource: {exc.detail}"},
 )

@router.post(“/”)
def create_item(item: Dict):
“””
新しい商品を作成します。カスタム例外を発生させる可能性のある例。
“””
item_name = item.get(“name”)
if not item_name:
# 名前のない商品は作成エラーとする (例)
raise ItemCreationError(item_name=”[Name not provided]”) # <– カスタム例外を発生

item_id = item_name.lower().replace(" ", "_")
if item_id in fake_items_db:
     # 既に存在する商品の場合は HTTPException を発生
     raise HTTPException(status_code=409, detail=f"Item '{item_name}' already exists")

fake_items_db[item_id] = item
return {"message": "Item created successfully", "item_id": item_id, "item": item}

… 他のエンドポイント …

“`

この例では、items ルーター内で ItemCreationError というカスタム例外が発生した場合に、定義した item_creation_exception_handler が実行され、カスタムのJSONレスポンスが返されます。また、ルーターレベルで HTTPException のハンドラーを定義することで、このルーター内で発生した HTTPException をグローバルなハンドラーよりも優先して処理できます。

ハンドラーの優先順位

FastAPIの例外ハンドラーは、以下の順で評価されます。

  1. パスオペレーション固有の exception_handler (デコレーター): 特定のパスオペレーション関数に @router.exception_handler() や @app.exception_handler() のようにデコレーターとして直接適用されたハンドラー（これはあまり一般的ではありません）。
  2. ルーター固有の add_exception_handler(): router.add_exception_handler() で登録されたハンドラー。登録された順に評価されます。
  3. アプリケーション全体の add_exception_handler(): app.add_exception_handler() で登録されたハンドラー。登録された順に評価されます。
  4. FastAPIのデフォルトハンドラー: HTTPException, RequestValidationException など、FastAPIが標準で提供するハンドラー。

ルーターレベルでハンドラーを設定することで、特定の機能グループで発生する可能性のあるエラーに対して、よりきめ細かい制御やカスタムレスポンスを提供できるようになります。

ミドルウェアとAPIRouter

ミドルウェアは、全てのリクエストが処理される前（または後）に実行される関数です。認証、ロギング、CORS処理、セッション管理など、様々な横断的関心事を処理するために使用されます。

FastAPIでは、ミドルウェアは通常アプリケーション全体に対して設定します。これは main.py で app.add_middleware() を使用して行います。

“`python

app/main.py (ミドルウェアの例)

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
import time

from .routers import all_routers

app = FastAPI(
title=”My Awesome API”,
description=”This is a sample API using FastAPI with APIRouter.”,
version=”0.1.0″,
)

CORSミドルウェアをアプリケーション全体に適用

app.add_middleware(
CORSMiddleware,
allow_origins=[““], # 必要に応じて制限
allow_credentials=True,
allow_methods=[““],
allow_headers=[“*”],
)

カスタムミドルウェアをアプリケーション全体に適用

class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() – start_time
response.headers[“X-Process-Time”] = str(process_time)
print(f”Request processed in {process_time:.4f} seconds”)
return response

app.add_middleware(TimingMiddleware)

ルーターの登録 (これまでと同じ)

for router in all_routers:
app.include_router(router)

@app.get(“/”)
def read_root():
return {“message”: “Welcome to My Awesome API!”}
“`

ミドルウェアは APIRouter ごとに設定することも可能ですが、これはあまり一般的ではありません。APIRouter のコンストラクタに middleware パラメータはありません。ルーター固有のミドルウェアを設定したい場合は、ルーターインスタンスに対して router.add_middleware() を使用します。

“`python

app/routers/items.py (ルーター固有ミドルウェアの例 – あまり使われない)

from fastapi import APIRouter, Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse
import time

router = APIRouter(
prefix=”/items”,
tags=[“items”],
responses={404: {“description”: “Item not found”}},
)

ルーター固有のカスタムミドルウェア

class ItemSpecificMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
print(“— ItemSpecificMiddleware START —“)
response = await call_next(request)
response.headers[“X-Item-Specific”] = “true”
print(“— ItemSpecificMiddleware END —“)
return response

ルーターにミドルウェアを追加

これは include_router する前に実行する必要がある

router.add_middleware(ItemSpecificMiddleware)

@router.get(“/”)
def read_items():
“””
ミドルウェアが適用されるエンドポイント。
“””
return {“message”: “Reading items”}

… 他のエンドポイント …

“`

main.py でルーターを include_router() する前に router.add_middleware() を呼び出す必要があります。

“`python

app/main.py (ルーター固有ミドルウェアを含むルーターをインクルードする例)

from fastapi import FastAPI
from .routers import items # items ルーターのみインポート (カスタムミドルウェア設定のため)
from .routers import all_routers # 他のルーターはまとめてインポート

app = FastAPI(…) # アプリケーション全体ミドルウェアもここで設定

items ルーターをインクルードする前にミドルウェアを追加

(items.py 内で add_middleware を呼び出している場合はこのステップは不要だが、

どちらかの方法で設定する必要がある)

items.router.add_middleware(…) # items.py内で既に定義している場合はこれは不要

ルーターをアプリケーションに登録

items ルーターを include_router する

app.include_router(items.router)

all_routers から items ルーターを除いた他のルーターを include_router

例: all_routers = [users_router, items_router] の場合

app.include_router(users_router) # users ルーターをインクルード

または以下のようにフィルタリング

for router in all_routers:

if router != items.router:

app.include_router(router)

@app.get(“/”)
def read_root():
return {“message”: “Welcome to My Awesome API!”}
“`

注意点: APIRouter にミドルウェアを適用することは可能ですが、一般的にはミドルウェアはアプリケーション全体に適用することが多いです。ルーター固有のミドルウェアが必要になるユースケースは限られます。例えば、特定のルーターに対してのみ厳格なレートリミットを適用したい場合などが考えられます。しかし、多くの場合、ルーターレベルの dependencies を使用する方が、特定のルーターのエンドポイント群に共通の処理を適用する目的には適しています。

テスト方法

FastAPIアプリケーションのテストには、fastapi.testclient.TestClient を使用するのが一般的です。これはASGIアプリケーション（FastAPIアプリケーションインスタンス）に対して同期的にリクエストを送信できるクライアントです。

APIRouter を使用してアプリケーションを分割した場合でも、通常は main.py で全てが統合されたFastAPIアプリケーションインスタンスに対してテストを行います。これにより、ルーター間の連携やアプリケーション全体の設定（ミドルウェア、例外ハンドラーなど）も含めて、実際の実行に近い形でテストできます。

“`python

tests/test_main.py

from fastapi.testclient import TestClient
from app.main import app # 統合されたFastAPIアプリケーションインスタンスをインポート

client = TestClient(app)

def test_read_root():
response = client.get(“/”)
assert response.status_code == 200
assert response.json() == {“message”: “Welcome to My Awesome API!”}

def test_read_users():
response = client.get(“/users/”)
assert response.status_code == 200
# ダミーデータに基づく期待値
assert len(response.json()) >= 0 # ダミーデータは変動するため件数は柔軟にチェック

def test_read_user_exists():
# ダミーデータに存在するユーザーIDでテスト
response = client.get(“/users/1”)
assert response.status_code == 200
assert response.json()[“username”] == “john_doe”

def test_read_user_not_exists():
# 存在しないユーザーIDでテスト
response = client.get(“/users/999”)
assert response.status_code == 404
assert response.json() == {“detail”: “User not found”} # ルーターレベルの responses にも一致する

def test_create_user():
new_user_data = {“username”: “test_user”, “full_name”: “Test User”}
response = client.post(“/users/”, json=new_user_data)
assert response.status_code == 200 # ダミー実装では成功を返す
assert “user_id” in response.json()
assert response.json()[“user”][“username”] == “test_user”

def test_read_items():
response = client.get(“/items/”)
assert response.status_code == 200
assert isinstance(response.json(), dict) # ダミーデータは辞書

def test_read_item_exists():
response = client.get(“/items/foo”)
assert response.status_code == 200
assert response.json()[“name”] == “Foo”

def test_read_item_not_exists():
response = client.get(“/items/nonexistent”)
assert response.status_code == 404
assert response.json() == {“detail”: “Item not found”} # ルーターレベルの responses にも一致する

def test_create_item_with_name():
new_item_data = {“name”: “New Item”, “price”: 100}
response = client.post(“/items/”, json=new_item_data)
# items.py のダミー実装では ItemCreationError になるケース
assert response.status_code == 200 # ダミー実装では成功を返す
assert “item_id” in response.json()
assert response.json()[“item”][“name”] == “New Item”

def test_create_item_without_name_triggers_error():
new_item_data = {“price”: 100} # 名前なし
response = client.post(“/items/”, json=new_item_data)
# items.py で ItemCreationError を発生させている
assert response.status_code == 400 # カスタム例外ハンドラーで設定したステータスコード
assert response.json() == {“message”: “Failed to create item: [Name not provided]”}
“`

この例では、main.py で app インスタンスをインポートし、TestClient に渡しています。これにより、分割されたルーターが全て登録された状態のアプリケーション全体をテストできます。

依存関係のモック

サービス層やデータベース接続などの依存関係を持つエンドポイントをテストする場合、実際の依存関係（例: データベースへの接続）をモック（Mock）して、テストの実行速度を上げたり、外部環境への依存をなくしたりすることが推奨されます。

FastAPIのDIシステムでは、app.dependency_overrides 属性を使うことで、特定の依存性プロバイダー関数（Depends() で使用される関数）を一時的に別の関数に置き換えることができます。これはテスト時に特に便利です。

“`python

app/services/user_service.py の UserService と get_user_service を使用

tests/test_users.py (サービス層をモックする例)

from fastapi.testclient import TestClient
from app.main import app
from app.routers.users import router # router インスタンスを直接インポート
from app.services.user_service import UserService, get_user_service
from typing import List, Dict

client = TestClient(app)

テスト用のモックUserServiceクラス

class MockUserService:
def get_users(self) -> List[Dict]:
print(“MockUserService: get_users 実行”)
# テスト用のダミーデータを返す
return [
{“username”: “mock_user_1”, “full_name”: “Mock User 1”},
{“username”: “mock_user_2”, “full_name”: “Mock User 2”},
]

def get_user_by_id(self, user_id: str) -> Dict | None:
    print(f"MockUserService: get_user_by_id({user_id}) 実行")
    # テスト用のダミーデータを返す (IDに基づく簡易的なモック)
    if user_id == "mock_id":
        return {"username": "mock_user", "full_name": "Mock User"}
    return None

def create_user(self, user_data: Dict) -> Dict:
    print("MockUserService: create_user 実行")
    # 作成成功を模倣
    return {"user_id": "mock_created_id", **user_data}

テスト用の依存性プロバイダー関数

def override_get_user_service():
“””
get_user_service 依存性を MockUserService で置き換える関数。
“””
print(“override_get_user_service 実行”)
# ここでは DB セッションを使わない MockUserService を返す
return MockUserService()

テスト関数

def test_read_users_mocked_service():
# app.dependency_overrides に置き換えたい依存性プロバイダーを指定
app.dependency_overrides[get_user_service] = override_get_user_service

try:
    response = client.get("/users/")
    assert response.status_code == 200
    # モックサービスが返すデータと一致するか確認
    assert response.json() == [
         {"username": "mock_user_1", "full_name": "Mock User 1"},
         {"username": "mock_user_2", "full_name": "Mock User 2"},
    ]
finally:
    # テスト終了後、依存性の置き換えを解除することを忘れない
    app.dependency_overrides.clear() # 全て解除
    # または個別に解除: del app.dependency_overrides[get_user_service]

def test_read_user_exists_mocked_service():
app.dependency_overrides[get_user_service] = override_get_user_service
try:
response = client.get(“/users/mock_id”)
assert response.status_code == 200
assert response.json() == {“username”: “mock_user”, “full_name”: “Mock User”}
finally:
app.dependency_overrides.clear()

def test_read_user_not_exists_mocked_service():
app.dependency_overrides[get_user_service] = override_get_user_service
try:
response = client.get(“/users/nonexistent_id”)
assert response.status_code == 404 # サービスが None を返した場合、ルーターが 404 を返すことを確認
assert response.json() == {“detail”: “User not found”}
finally:
app.dependency_overrides.clear()
“`

このテストでは、get_user_service 依存性を、データベースに依存しない override_get_user_service 関数で一時的に置き換えています。override_get_user_service は、テスト用に用意した MockUserService のインスタンスを返します。これにより、ルーターが正しく UserService のメソッドを呼び出しているか、サービスが返した値に応じて適切なレスポンスを返しているかなど、サービス層との連携部分をテストできます。

app.dependency_overrides を使うことで、分割されたルーターが依存している他のコンポーネント（サービス、DB、外部APIクライアントなど）を簡単にモックでき、ルーター単体のロジック（ルーティング、パラメータ検証、依存性解決、サービスの呼び出し、レスポンス生成）に焦点を当てたテストが可能になります。

注意点: app.dependency_overrides.clear() または del app.dependency_overrides[dependency_provider] を使って、テストの終了後に依存性の置き換えを解除することを忘れないでください。これを怠ると、他のテストに影響を与える可能性があります。理想的には、pytestの yield_fixture を使うか、各テスト関数で try...finally ブロックを使用するのが良いでしょう。

大規模開発におけるAPIRouterの活用

APIRouter は、小規模なAPIだけでなく、大規模で複雑なアプリケーションやマイクロサービスアーキテクチャにおいても非常に強力なツールです。

バージョン管理

APIのバージョン管理は、後方互換性を維持しながらAPIに変更を加えるためによく行われます。一般的なバージョン管理の方法として、パスにバージョン番号を含める方法があります（例: /v1/users, /v2/users）。APIRouter を使うと、このバージョン管理を非常にきれいに実現できます。

“`python

app/routers/v1/users.py

from fastapi import APIRouter

v1 ユーザーAPIルーター

router = APIRouter(
prefix=”/v1/users”, # <– v1 のパスプレフィックス
tags=[“users (v1)”],
)

@router.get(“/”)
def read_users_v1():
return [{“id”: 1, “name”: “User V1 A”}, {“id”: 2, “name”: “User V1 B”}]

@router.get(“/{user_id}”)
def read_user_v1(user_id: int):
return {“id”: user_id, “name”: f”User V1 {user_id}”}
“`

“`python

app/routers/v2/users.py

from fastapi import APIRouter

v2 ではユーザーモデルに email を追加したとする (Pydantic モデルを使用するのがより現実的)

from …models import UserV2 # 実際のコードではモデル定義が必要

v2 ユーザーAPIルーター

router = APIRouter(
prefix=”/v2/users”, # <– v2 のパスプレフィックス
tags=[“users (v2)”],
)

@router.get(“/”)
def read_users_v2():
# v2 では email も返すようになった (という想定)
return [
{“id”: 1, “name”: “User V2 A”, “email”: “[email protected]”},
{“id”: 2, “name”: “User V2 B”, “email”: “[email protected]”},
]

@router.get(“/{user_id}”)
def read_user_v2(user_id: int):
# v2 では email も返すようになった (という想定)
return {“id”: user_id, “name”: f”User V2 {user_id}”, “email”: f”{user_id}@example.com”}

v2 で追加されたエンドポイント (例)

@router.post(“/”)
def create_user_v2(user_data: dict): # 実際は UserV2 モデルを使用
# v2 のユーザー作成ロジック
print(f”Creating user V2: {user_data}”)
return {“message”: “User V2 created”, “user”: user_data}
“`

“`python

app/main.py (バージョン別ルーターの登録)

from fastapi import FastAPI

v1 および v2 のルーターをインポート

(routers/init.py でまとめて管理することも可能)

from .routers.v1 import users as users_v1
from .routers.v2 import users as users_v2
from .routers import items # items ルーターはバージョン管理しないと仮定

app = FastAPI(
title=”Versioned API Example”,
version=”1.0.0″, # アプリケーション全体のバージョン (通常は最新バージョンに合わせるか別で管理)
)

バージョン別ルーターを登録

app.include_router(users_v1.router)
app.include_router(users_v2.router)
app.include_router(items.router) # バージョン管理しないルーターも登録

@app.get(“/”)
def read_root():
return {“message”: “Welcome to the Versioned API!”}
“`

この構成により、/v1/users と /v2/users の両方のエンドポイントを同時に提供できます。クライアントは必要なバージョンのパスを指定することで、目的のAPIを利用できます。新しいバージョンを開発する際は、既存のバージョンに影響を与えずに新しいルーターファイルを作成・編集できます。

チーム開発とコード分担

大規模なプロジェクトでは、複数の開発者が同時に異なる機能に取り組むのが一般的です。APIRouter を使ってコードを機能ごとに分割することで、開発者は他の機能のコードにほとんど触れることなく、自分が担当する機能に関連するルーターファイルやサービスファイルに集中できます。これにより、コードのコンフリクトが減少し、チーム開発の効率が向上します。

マイクロサービスアーキテクチャ

APIRouter は、マイクロサービスアーキテクチャを構築する際の基盤としても利用できます。各マイクロサービスは、特定の機能ドメイン（例: ユーザーサービス、商品サービス、注文サービス）を担当し、それぞれのサービスが独自のFastAPIアプリケーションとして稼働します。各マイクロサービス内のAPIは、さらに APIRouter を使って詳細に分割されることもあります。サービス間の通信はHTTPやメッセージキューなどで行われます。

ただし、APIRouter 自体がマイクロサービス化を直接行うわけではありません。APIRouter はあくまで一つのFastAPIアプリケーション内でのコード分割・整理のためのツールです。

APIRouter使用上の注意点とベストプラクティス

APIRouter は強力なツールですが、適切に使用しないと逆にコードが分かりづらくなることもあります。以下に、使用上の注意点とベストプラクティスをいくつか挙げます。

  1. ルーターの粒度を適切に保つ:
    • ルーターを細かく分けすぎると、ファイルの数が増えすぎて管理が大変になります。
    • 逆に、一つのルーターに多くの無関係なエンドポイントを含めると、分割の意味が薄れます。
    • 一般的には、リソース（ユーザー、商品、注文など）ごと、または主要な機能領域ごとにルーターを分割するのが分かりやすいでしょう。
  2. 一貫性のあるパス命名規則:
    • prefix を活用し、ルーターごとに一貫したパス構造を適用してください。例えば、全てのリソースは複数形 (/users, /items) を使うなど。
    • バージョン管理を行う場合は、明確なバージョニングスキーム（例: /v1/users, /v2/items）を採用してください。
  3. tags の適切な利用:
    • tags を使うとドキュメントが劇的に見やすくなります。機能やリソースに基づいて分かりやすいタグ名を付けてください。
    • 一つのエンドポイントに複数のタグを付けることも可能ですが、付けすぎるとかえって混乱を招くことがあります。
  4. 依存性の配置場所:
    • 共通の認証や権限チェックなど、ルーター内の 全て のエンドポイントに適用される依存性は、ルーターの dependencies パラメータで指定するのが適切です。
    • 特定のエンドポイント のみ に必要な依存性（例: データベースセッションを引数として受け取る、特定のクエリパラメータを検証するなど）は、そのパスオペレーション関数の引数で Depends() を使って指定してください。
  5. エラーハンドリングの一元化:
    • アプリケーション全体で共通のエラーハンドリング（例: 未知の例外、特定のHTTPエラーに対するカスタムレスポンスフォーマット）は、main.py で app.add_exception_handler() を使って設定するのが良いでしょう。
    • 特定のルーター内でのみ発生するカスタム例外や、そのルーター特有のHTTPエラー処理が必要な場合にのみ、ルーターレベルの router.add_exception_handler() を検討してください。
  6. Pydanticモデルの活用:
    • リクエストボディやレスポンスデータの構造は、Pydanticモデルを使って明確に定義し、models.py などのファイルにまとめてください。ルーターやサービスはこれらのモデルを参照するようにします。
  7. サービス層へのロジック分離:
    • ビジネスロジック（データベース操作、外部サービス呼び出し、計算など）はルーターから分離し、サービス層に記述してください。ルーターはリクエストを受け付けてサービスを呼び出すことに専念します。これはDIと組み合わせて実現します。
  8. __init__.py を活用したルーター管理:
    • app/routers/__init__.py でルーターをインポートし、リスト化してエクスポートする方法は、main.py でルーターをまとめて登録する際に便利です。ファイルが増えても main.py をシンプルに保てます。
  9. ドキュメント生成の確認:
    • ルーターを分割・設定したら、忘れずに /docs や /redoc にアクセスして、APIドキュメントが意図通りに生成されているか（パス、タグ、レスポンスなどが正しく表示されているか）を確認してください。

これらのベストプラクティスを実践することで、APIRouter を使ったFastAPIアプリケーションは、より構造化され、保守しやすく、チームで開発しやすいものになります。

まとめ

この記事では、FastAPIアプリケーションを整理・分割するための中心的な機能である APIRouter について、その基本的な使い方から詳細設定、プロジェクト構造、依存性の注入、エラーハンドリング、テスト、大規模開発での応用まで、幅広く深く掘り下げて解説しました。

APIRouter を使用することで、APIエンドポイントを機能別やリソース別に独立したファイルやモジュールとして管理できるようになります。これにより、コードの可読性が向上し、保守が容易になり、特に複数の開発者が関わるチーム開発において、コードの競合を減らし、効率を大幅に向上させることができます。

また、APIRouter の prefix や tags といったパラメータは、APIの構造を設計する上で役立つだけでなく、FastAPIが自動生成するOpenAPIドキュメントの質を高め、APIの利用者がエンドポイントを理解しやすくなるという副次的な効果ももたらします。

さらに、FastAPIの強力な依存性注入システムと APIRouter を組み合わせることで、ルーター層からビジネスロジックをサービス層へ適切に分離し、よりテストしやすく、再利用可能なコード構造を構築することが可能になります。

アプリケーションの規模が大きくなるにつれて、適切なコードの分割と整理はプロジェクト成功の鍵となります。FastAPIの APIRouter は、この目的を達成するための強力なツールです。

この記事が、あなたのFastAPI開発において APIRouter を効果的に活用し、より良いAPIアプリケーションを構築するための一助となれば幸いです。

FastAPIには、ここで紹介した機能以外にも、認証、データベース連携（SQLAlchemy, Pydantic-SQLAlchemy, Ormliteなど）、非同期タスク、WebSocketなど、様々な強力な機能があります。これらの機能を APIRouter で分割・整理された構造の中で活用していくことで、さらに複雑で高機能なAPIアプリケーションを構築していくことができるでしょう。

Happy Coding!

コメントする

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

上部へスクロール