【FastAPI】depends活用ガイド：共通処理・効率化の秘訣
はじめに：なぜFastAPIとdependsなのか？
Web API開発フレームワークとして、近年FastAPIの人気が急速に高まっています。その理由の一つに、高いパフォーマンス、直感的でモダンなPythonの型ヒントを活用した設計、そして自動生成されるAPIドキュメント（OpenAPI、Swagger UI、ReDoc）の存在があります。しかし、FastAPIが開発者の生産性とコードの保守性を劇的に向上させている最大の要因の一つは、その強力な依存関係注入（Dependency Injection, DI）システム、特にdependsの仕組みにあります。
API開発において、認証、データベース接続、設定値の読み込み、入力データのバリデーション、共通的なロジックの実行など、多くの処理が複数のエンドポイント間で共通して必要となります。これらの共通処理を各エンドポイント関数内に直接記述すると、コードが重複し、読みにくく、テストしにくく、保守性が著しく低下します。
ここで登場するのがFastAPIのdependsです。dependsを使用することで、これらの共通処理や事前準備のロジックを独立した関数（またはクラス）として定義し、それをAPIエンドポイント関数が必要に応じて「依存」する形で注入することができます。これにより、エンドポイント関数は自身の本来の責務（リクエスト処理とレスポンス生成）に集中できるようになり、コードのモジュール化、再利用性、テスト容易性、そして全体の効率化が実現されます。
この記事では、FastAPIのdependsを徹底的に掘り下げ、その基本的な使い方から、認証、データベース管理といった共通処理への適用、さらにはコードの効率化、テストの容易化といった高度な活用方法までを、豊富なコード例を交えながら詳細に解説します。この記事を読み終える頃には、あなたはFastAPIのdependsを自在に操り、よりクリーンで効率的、かつ保守性の高いAPIを開発できるようになっているでしょう。
dependsの基本：依存関係注入とは？
dependsを理解するためには、まず「依存関係注入（Dependency Injection, DI）」というソフトウェア設計の原則について軽く触れておく必要があります。
依存関係注入とは、あるオブジェクトや関数が、自身が機能するために必要とする他のオブジェクトや関数（＝依存関係）を、自身の内部で生成したり取得したりするのではなく、外部から渡されるように設計することです。これにより、オブジェクトや関数は特定の具象的な依存関係に直接結びつくことなく、より汎用的でテストしやすいコードになります。
例えば、データベースにアクセスする必要がある関数があるとします。DIを使用しない場合、その関数内で直接データベース接続を確立したり、グローバルな接続オブジェクトに依存したりすることが考えられます。
“`python
DIを使用しない例（非推奨）
import database
def get_user(user_id: int):
db = database.connect() # データベース接続を関数内で直接生成
user = db.query(“SELECT * FROM users WHERE id = ?”, user_id)
db.close()
return user
“`
このコードは、get_user関数がデータベース接続の詳細に直接依存しています。これは以下の問題を引き起こします。
- テストの困難さ: データベースに実際に接続しないとこの関数をテストできません。ユニットテストでデータベースアクセス部分をモックするのが難しい場合があります。
- 柔軟性の低さ: 別のデータベースシステムに切り替えたい場合、この関数を含め、接続処理を直接書いているすべての場所を変更する必要があります。
- リソース管理: 接続の生成とクローズのロジックが各所に散らばり、管理が煩雑になります。
DIの考え方では、get_user関数は必要なデータベース接続オブジェクトを外部から受け取るように設計します。
“`python
DIの考え方に基づいた例
実際には接続プールなどを使うことが多い
class DatabaseConnection:
def query(self, sql, *params):
# ダミーのクエリ実行
print(f”Executing query: {sql} with params {params}”)
return {“id”: params[0], “name”: “Test User”}
def get_user(user_id: int, db: DatabaseConnection): # 依存関係を引数として受け取る
user = db.query(“SELECT * FROM users WHERE id = ?”, user_id)
return user
使用例
db_conn = DatabaseConnection() # 依存関係は外部で生成
user_data = get_user(123, db_conn) # 依存関係を渡す
“`
この例では、get_user関数はDatabaseConnection型のオブジェクトを引数として受け取ります。これにより、get_user関数は「どのようなデータベース接続オブジェクトを受け取るか」には関心がなく、「与えられたオブジェクトを使ってクエリを実行できるか」だけに関心を持ちます。
FastAPIのdependsは、このDIの原則をWeb APIのエンドポイント関数に対して非常に容易に実現するための仕組みです。エンドポイント関数の引数として、Depends()でラップされた依存関係を指定することで、FastAPIはその依存関係を解決し、結果を関数の引数に渡してくれます。
dependsの構文と基本的な使い方
dependsは、fastapiモジュールからインポートして使用します。基本的な構文は以下の通りです。
“`python
from fastapi import Depends, FastAPI, HTTPException
app = FastAPI()
依存関係となる関数
def get_query_param(param: str):
“””クエリパラメータを受け取り、そのまま返す依存関係”””
print(f”Dependency called with param: {param}”)
return param
@app.get(“/items/”)
def read_items(
# エンドポイント関数の引数として、Depends()でラップして依存関係を指定
query_param_value: str = Depends(get_query_param)
):
“””
query_param_value には get_query_param 関数の返り値が注入される
“””
return {“query_param_from_dependency”: query_param_value}
http://localhost:8000/items/?param=test でアクセス
出力:
Dependency called with param: test
レスポンス:
{“query_param_from_dependency”:”test”}
“`
この例では、read_itemsエンドポイント関数はquery_param_valueという引数を持ち、そのデフォルト値としてDepends(get_query_param)が指定されています。これはFastAPIに対して、「このエンドポイント関数を実行する前にget_query_param関数を実行し、その返り値をquery_param_value引数に渡してください」と指示しています。
get_query_param関数は、自身の引数としてparam: strを受け取っています。FastAPIは、この依存関係関数が必要とする引数も自動的に解決しようとします。この例の場合、param: strはリクエストのクエリパラメータparamに対応するため、FastAPIはリクエストからparamクエリパラメータの値を取得し、それをget_query_param関数に渡します。
依存関係関数（get_query_param）が実行され、その返り値がエンドポイント関数（read_items）の引数（query_param_value）に注入される、これがdependsの基本的な流れです。
依存関係関数は、リクエストパラメータ（パスパラメータ、クエリパラメータ、ヘッダー、クッキー、リクエストボディなど）を自身の引数として受け取ることができます。FastAPIはこれらの引数を自動的に解決し、依存関係関数に渡します。
依存関係でのバリデーションとエラーハンドリング
依存関係関数内で、リクエストパラメータのバリデーションや存在チェックを行い、問題があればHTTPExceptionを送出することができます。これは、エンドポイント関数本体でバリデーションを行うよりも、ロジックを分離する上で役立ちます。
“`python
from fastapi import Depends, FastAPI, HTTPException, Query
app = FastAPI()
def check_item_id(item_id: int = Query(…, ge=1)):
“””item_id が1以上であることを確認し、item_id を返す依存関係”””
# Query() を使うことで、Swagger UIなどにもパラメータ情報が反映される
# ge=1 で1以上であることをバリデーション
# … は必須であることを示す
# さらに複雑なカスタムバリデーションも可能
if item_id > 1000:
raise HTTPException(status_code=400, detail="Item ID cannot exceed 1000")
print(f"Dependency check_item_id called with item_id: {item_id}")
return item_id
@app.get(“/items/{item_id}”)
def read_item(
# パスパラメータを依存関係に渡す
# FastAPIはパスパラメータ {item_id} の値を check_item_id の item_id 引数に自動で渡す
validated_item_id: int = Depends(check_item_id)
):
“””
依存関係によってバリデーションされた item_id を受け取る
“””
return {“item_id”: validated_item_id, “message”: “Item ID is valid”}
例:
GET /items/500 -> {“item_id”: 500, “message”: “Item ID is valid”}
GET /items/0 -> 422 Unprocessable Entity (Query(…, ge=1) によるバリデーションエラー)
GET /items/2000 -> 400 Bad Request (check_item_id 関数内のカスタムエラー)
“`
この例のように、依存関係関数内でfastapiが提供するQuery, Path, Header, Cookie, Bodyなどのパラメータ定義オブジェクトを使用できます。これらはPydanticのフィールド定義と同様の機能を提供し、自動的なデータ変換とバリデーションを行います。さらに、カスタムのバリデーションロジックを記述し、条件を満たさない場合にHTTPExceptionを発生させることで、エラーレスポンスを返すことができます。
依存関係内で発生したHTTPExceptionは、FastAPIによって自動的に適切なHTTPレスポンスに変換されます。
共通処理としてのdepends活用
API開発におけるdependsの最も強力な使い方の1つは、複数のエンドポイントで必要とされる共通処理をカプセル化することです。これにより、コードの重複を避け、保守性を大幅に向上させることができます。代表的な共通処理として、認証・認可、データベースセッション管理、設定値の読み込みなどがあります。
認証・認可処理
多くのAPIでは、ユーザーが認証されているか、あるいは特定のアクションを実行する権限（認可）を持っているかを確認する必要があります。この処理を各エンドポイントに記述するのは非効率的です。dependsを使えば、認証・認可ロジックを一度だけ定義し、必要なエンドポイントに依存関係として追加するだけで済みます。
FastAPIは、OAuth2やJWTのような一般的な認証フローをサポートするためのユーティリティクラスをfastapi.securityモジュールで提供しています。これらはdependsと組み合わせて使用することを想定しています。
例1：APIキー認証
APIキーをHTTPヘッダーまたはクエリパラメータで受け取り、有効性を確認するシンプルな認証。
“`python
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.security import APIKeyHeader
app = FastAPI()
APIキーを ‘X-API-Key’ ヘッダーから取得する設定
api_key_header = APIKeyHeader(name=”X-API-Key”, auto_error=False) # auto_error=False で手動エラー処理
ダミーの有効なAPIキーリスト
VALID_API_KEYS = {
“abcdef123456”,
“xyz789012345”,
}
依存関係：APIキーの認証
def verify_api_key(api_key: str = Security(api_key_header)):
“””
APIキーヘッダーからキーを取得し、有効性を確認する
“””
# Security() を使うことで、OpenAPIスキーマにセキュリティ情報が追加される
if api_key is None or api_key not in VALID_API_KEYS:
# APIキーが無効な場合は HTTPException を送出
raise HTTPException(
status_code=401,
detail=”Invalid or missing API Key”,
headers={“WWW-Authenticate”: “Bearer”}, # 認証失敗時のヘッダー
)
# 有効な場合は特に返り値は不要（またはユーザー情報などを返す）
print(f”API Key ‘{api_key}’ is valid.”)
return api_key # 必要に応じてユーザー情報などを返すことも可能
@app.get(“/protected/”)
def protected_route(
# このエンドポイントは verify_api_key に依存する
# verify_api_key が HTTPException を送出すると、この関数は実行されない
api_key: str = Depends(verify_api_key)
):
“””
APIキー認証が必要なエンドポイント
“””
return {“message”: “Access granted!”, “used_api_key”: api_key}
アクセス例：
curl http://localhost:8000/protected/ -H “X-API-Key: abcdef123456” -> 成功
curl http://localhost:8000/protected/ -H “X-API-Key: invalid_key” -> 401 Unauthorized
curl http://localhost:8000/protected/ -> 401 Unauthorized
“`
Security()はDepends()のサブクラスであり、OpenAPIスキーマにセキュリティスキームの情報を追加するために使用されます。認証ロジック自体はverify_api_key関数内にカプセル化されており、必要なエンドポイント関数はDepends(verify_api_key)を指定するだけで、認証済みであることを保証できます。
例2：OAuth2/JWT認証
より一般的なトークンベースの認証では、JWT (JSON Web Token) がよく使用されます。ユーザー名/パスワードでログインしてJWTを取得し、以降のリクエストではAuthorizationヘッダーにそのJWTを添付します。
FastAPIはfastapi.security.oauth2モジュールでOAuth2関連のクラスを提供しており、これをJWT認証に応用できます。
“`python
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import BaseModel
from typing import Optional
ダミーのユーザーDB (実際はデータベースなどから取得)
fake_users_db = {
“johndoe”: {
“username”: “johndoe”,
“hashed_password”: “fakehashedpassword”, # 実際はハッシュ化されたパスワード
“full_name”: “John Doe”,
“email”: “[email protected]”,
},
“janedoe”: {
“username”: “janedoe”,
“hashed_password”: “fakehashedpassword2”,
“full_name”: “Jane Doe”,
“email”: “[email protected]”,
}
}
JWT設定 (実際は環境変数などから読み込む)
SECRET_KEY = “your-secret-key” # 署名に使用する秘密鍵
ALGORITHM = “HS256”
ACCESS_TOKEN_EXPIRE_MINUTES = 30 # トークン有効期限
トークン取得のエンドポイントURLを指定
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=”token”)
JWTペイロードのデータモデル (Pydantic)
class TokenData(BaseModel):
username: Optional[str] = None
依存関係：JWTトークンの検証
def get_current_user(token: str = Depends(oauth2_scheme)):
“””
Authorization ヘッダーから JWT トークンを取得し、検証してユーザーを返す
“””
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Could not validate credentials”,
headers={“WWW-Authenticate”: “Bearer”},
)
try:
# JWTをデコードしてペイロードを取得
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get(“sub”) # ‘sub’ クレームにユーザー名を入れるのが慣習
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
# JWTが無効な場合
raise credentials_exception
# ユーザー名を基にユーザー情報を取得 (ここではダミーDBから)
user = fake_users_db.get(token_data.username)
if user is None:
raise credentials_exception # ユーザーが存在しない場合も認証失敗
print(f"User '{user['username']}' authenticated via JWT.")
return user # 認証されたユーザー情報を返す
app = FastAPI()
ログイン（トークン発行）エンドポイント
@app.post(“/token”)
def login_for_access_token(form_data: OAuth2PasswordBearer = Depends(oauth2_scheme)): # この例ではトークン生成なのでこの dépendence は不要です。正しくは OAuth2PasswordRequestForm ですが、話を単純にするため省略
# 実際はユーザー名/パスワードを受け取り、検証してJWTを生成
# ここでは簡易的に、入力されたトークンを検証してユーザーを返す例にする
# …本来はここでユーザー名/パスワード検証とトークン生成ロジックが入る…
# 簡易的に、ダミーDBにユーザーがいれば成功とする
user = fake_users_db.get(form_data.username if hasattr(form_data, ‘username’) else ‘johndoe’) # 仮のユーザー取得
if not user: # or 実際はパスワード不一致など
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=”Incorrect username or password”,
headers={“WWW-Authenticate”: “Bearer”},
)
# … JWT生成ロジック …
# access_token = create_access_token(data={“sub”: user[‘username’]}, expires_delta=access_token_expires)
# return {“access_token”: access_token, “token_type”: “bearer”}
# この例では検証済みのユーザー情報をそのまま返すことにします
print(f”User ‘{user[‘username’]}’ logged in.”)
return user # 実際はトークンを返す
認証が必要なエンドポイント
@app.get(“/users/me/”)
def read_users_me(current_user: dict = Depends(get_current_user)):
“””
ログインユーザー情報を取得
“””
return current_user
アクセス例：
1. ログインエンドポイントでトークンを取得 (ここでは省略。通常はPOSTでユーザー名/パスワードを送信)
2. 取得したトークンを使って /users/me/ にアクセス
curl http://localhost:8000/users/me/ -H “Authorization: Bearer eyJhbGciOiJIUzI1NiIsIn…” (取得したJWT) -> 成功
curl http://localhost:8000/users/me/ -> 401 Unauthorized
“`
get_current_user関数は、Depends(oauth2_scheme)によってHTTPリクエストからJWTトークンを取得し、joseライブラリなどを使ってそのトークンを検証します。トークンが無効であればHTTPExceptionを送出し、有効であればトークンから抽出した情報（例：ユーザー名）を基にデータベースなどからユーザー情報を取得して返します。
read_users_meエンドポイントはDepends(get_current_user)を指定することで、この関数が実行されることを保証し、認証済みのユーザー情報をcurrent_user引数として受け取ることができます。エンドポイント関数自身は認証のロジックについて何も知る必要がなく、受け取ったcurrent_userを使って処理を進めるだけです。
データベースセッション管理
Webアプリケーションにおいて、リクエストごとにデータベースセッションを作成し、処理が完了したらコミットまたはロールバックしてセッションを閉じる、というパターンは非常に一般的です。この処理もdependsを使って共通化できます。特に、Pythonのyieldキーワードを使った「Generator Dependencies」がこの用途に最適です。
yieldを使った依存関係では、yieldキーワードの前にあるコードがエンドポイント関数の実行前に実行され、yieldが返す値が依存関係の返り値として注入されます。そして、エンドポイント関数の実行が完了した後（正常終了、例外発生問わず）、yieldキーワードの後に続くコードが実行されます。これはリソースの後処理に非常に便利です。
例：SQLAlchemyを使ったデータベースセッション管理
“`python
from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base # ORMモデルの基底クラス
from . import models # 実際のアプリケーションでは models.py などに定義
from .database import SessionLocal, engine # 実際のアプリケーションでは database.py などに定義
ダミーのデータベース設定 (実際は適切に設定)
SQLALCHEMY_DATABASE_URL = “sqlite:///./test.db” # SQLite例
SQLAlchemy エンジン作成
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={“check_same_thread”: False} # SQLiteのみ必要な設定
)
セッションファクトリ作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
ORMモデルの基底クラス
Base = declarative_base()
ダミーのORMモデル (実際は適切に定義)
class Item(Base):
tablename = “items”
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
(ここではモデル定義は省略し、セッション取得のみに焦点を当てる)
DB接続/セッション取得の依存関係 (yield を使用)
def get_db():
“””
リクエストごとにDBセッションを作成し、終了時にクローズする
“””
db = SessionLocal() # セッション作成
print(“Database session created.”)
try:
yield db # セッションをエンドポイント関数に注入
# エンドポイント関数の実行が完了した後、ここから下のコードが実行される
print(“Database session committed.”)
db.commit() # 変更をコミット
except Exception as e:
# 例外が発生した場合、ロールバック
print(f”Database session rolled back due to exception: {e}”)
db.rollback()
raise # 例外を再送出
finally:
# 処理が正常終了または例外終了したかにかかわらず、セッションをクローズ
print(“Database session closed.”)
db.close()
app = FastAPI()
@app.get(“/items/”)
def read_items(db: Session = Depends(get_db)):
“””
DBセッションを使ってアイテム一覧を取得する (ダミー実装)
“””
# db は get_db から yield された Session オブジェクト
# ここで SQLAlchemy のセッションを使ってDB操作を行う
# 例: items = db.query(models.Item).all()
print(“Accessing DB session within endpoint.”)
return {“message”: “DB session accessed”, “session_type”: str(type(db))}
@app.post(“/items/”)
def create_item(item_data: dict, db: Session = Depends(get_db)):
“””
DBセッションを使って新しいアイテムを作成する (ダミー実装)
“””
# item_data を使って新しいレコードを作成し、db.add(), db.commit(), db.refresh() などを行う
print(f”Creating item: {item_data}”)
# 例外を発生させてロールバックの挙動を確認
if “error” in item_data:
raise HTTPException(status_code=400, detail=”Simulating DB error”)
print("Committing DB transaction.")
# ダミー処理: 実際はここで db.add(), db.commit() などを行う
return {"message": "Item created (simulated)", "received_data": item_data}
アクセス例：
GET http://localhost:8000/items/
-> Database session created.
Accessing DB session within endpoint.
Database session committed.
Database session closed.
{“message”: “DB session accessed”, “session_type”: ““}
POST http://localhost:8000/items/ with JSON body {“name”: “New Item”}
-> Database session created.
Creating item: {‘name’: ‘New Item’}
Committing DB transaction.
Database session committed.
Database session closed.
{“message”: “Item created (simulated)”, “received_data”: {“name”: “New Item”}}
POST http://localhost:8000/items/ with JSON body {“name”: “Error Item”, “error”: true}
-> Database session created.
Creating item: {‘name’: ‘Error Item’, ‘error’: True}
Database session rolled back due to exception: Simulated DB error
Database session closed.
400 Bad Request レスポンス
“`
get_db関数は、リクエスト開始時に新しいSQLAlchemyセッションを作成し、yield dbでそのセッションをエンドポイント関数に渡します。エンドポイント関数が正常に完了すると、yieldの後のdb.commit()が実行され、変更がコミットされます。例外が発生した場合はexceptブロックが実行され、db.rollback()が行われて変更が取り消され、例外が再送出されます。finallyブロックは常に実行され、セッションが閉じられます。
このパターンにより、各エンドポイント関数は単にdb: Session = Depends(get_db)と宣言するだけで、自動的に適切なライフサイクルのデータベースセッションを受け取ることができます。コミットやロールバック、クローズといった煩雑なセッション管理コードを各エンドポイントから排除し、一元管理できるため、コードが非常にクリーンで保守しやすくなります。
設定値の読み込み
アプリケーションの設定値（データベース接続文字列、APIキー、外部サービスURLなど）は、通常、起動時に一度だけ読み込み、アプリケーション全体で利用できるようにします。これらの設定値をエンドポイント関数に直接渡すのではなく、依存関係として提供することも可能です。
“`python
from fastapi import Depends, FastAPI
from pydantic import BaseSettings # pydantic-settings を使用する場合
Pydantic-settings を使用した設定モデル (推奨)
pip install pydantic-settings
class Settings(BaseSettings):
database_url: str = “sqlite:///./default.db”
api_key: str = “default_api_key”
class Config:
env_file = “.env” # .env ファイルから設定を読み込む
設定オブジェクトを取得する依存関係
この関数は一度だけ実行され、その結果がキャッシュされる
def get_settings():
print(“Loading settings…”)
# ここでファイルから読み込む、環境変数を処理するなどを行う
# Pydantic BaseSettings は環境変数や .env ファイルから自動で読み込む
return Settings()
FastAPI インスタンス作成
app = FastAPI()
設定オブジェクトを依存関係として注入
@app.get(“/info/”)
def read_info(settings: Settings = Depends(get_settings)):
“””
設定情報の一部を表示するエンドポイント
“””
# settings は get_settings() の返り値 (Settings オブジェクト)
print(“Accessing settings within endpoint.”)
return {
“database_url”: settings.database_url,
“api_key_prefix”: settings.api_key[:5] + “…”, # APIキー全体は表示しない
}
.env ファイルの例:
DATABASE_URL=”postgresql://user:password@host/db”
API_KEY=”your-secure-api-key-from-env”
実行例:
環境変数や .env ファイルに設定がなければデフォルト値を使用
環境変数 DATABASE_URL=”postgresql://user:password@host/db” が設定されていればその値を使用
GET http://localhost:8000/info/
-> Loading settings… (初回アクセス時のみ)
Accessing settings within endpoint.
{“database_url”: “…”, “api_key_prefix”: “your-… (または default_…)”}
“`
この例では、get_settings関数がSettingsオブジェクト（Pydantic BaseSettings を使用すると、環境変数や .env ファイルから設定を読み込んでくれる便利な機能）を生成します。get_settingsはFastAPIによって自動的にキャッシュされるため（デフォルトの挙動）、リクエストごとに設定を読み込み直すといった無駄は発生しません。設定オブジェクトが必要なエンドポイントは、settings: Settings = Depends(get_settings)と記述するだけで、その設定オブジェクトにアクセスできます。
効率化のためのdepends活用
dependsは共通処理の分離だけでなく、特定の処理フローを効率化するためにも使用できます。特に、リクエストパラメータに基づいたデータの事前取得や、複雑な入力バリデーションを依存関係に切り出すことが有効です。
入力データのバリデーションと変換
Pydanticモデルは強力な入力データバリデーションと変換を提供しますが、パスパラメータやクエリパラメータに対して、より複雑なバリデーションロジックやデータの変換が必要な場合があります。これをdependsで実現できます。
“`python
from fastapi import Depends, FastAPI, HTTPException, Path, Query
app = FastAPI()
def validate_and_get_item(item_id: int = Path(…, ge=1),
lang: str = Query(“en”, regex=”^(en|ja)$”)):
“””
パスパラメータ item_id と クエリパラメータ lang をバリデーションし、
加工したデータを返す依存関係
“””
# item_id は Path(…, ge=1) で 1 以上の整数であることを保証
# lang は Query(“en”, regex=”^(en|ja)$”) で ‘en’ または ‘ja’ であることを保証 (デフォルト値 ‘en’)
# 例：item_id を基に、複雑なロジックで関連データを取得するなど
# ここでは簡易的に、 item_id に応じて特別な処理を行う例
if item_id == 999:
raise HTTPException(status_code=404, detail="Special item not found")
# バリデーション済みの値をタプルや辞書として返す
# あるいは、item_id に基づいてデータベースから取得したオブジェクトを返すことが多い
print(f"Validated item_id: {item_id}, lang: {lang}")
return {"item_id": item_id, "language": lang}
@app.get(“/items/{item_id}”)
def read_item_with_validation(
validated_data: dict = Depends(validate_and_get_item)
):
“””
依存関係でバリデーション・取得されたデータを使用するエンドポイント
“””
# validated_data は validate_and_get_item の返り値
return {
“message”: “Item data processed”,
“item_id”: validated_data[“item_id”],
“language”: validated_data[“language”]
}
例：
GET /items/123?lang=en -> Validated item_id: 123, lang: en -> {“message”: …, “item_id”: 123, “language”: “en”}
GET /items/456?lang=fr -> Query() の regex バリデーションに失敗 -> 422 Unprocessable Entity
GET /items/999?lang=ja -> validate_and_get_item 内のカスタムエラー -> 404 Not Found
“`
この例では、validate_and_get_item依存関係が、item_idパスパラメータとlangクエリパラメータを受け取ります。PathやQueryオブジェクトを使って基本的な型チェックとバリデーションを行った上で、さらにカスタムロジック（例：item_id == 999の場合のエラー）を実行しています。依存関係の返り値は辞書ですが、データベースから取得したユーザーオブジェクトやアイテムオブジェクトなどを返すことも多いです。
エンドポイント関数は、バリデーションとデータ取得のロジックから切り離され、単に依存関係が提供した加工済みデータを受け取って利用するだけになります。
データのフェッチ（データベース、外部APIなど）
リクエストパラメータ（特にパスパラメータのID）に基づいて、特定のエンティティ（ユーザー、アイテムなど）をデータベースや外部APIから取得する必要がある場面は頻繁にあります。この「IDからオブジェクトを取得する」処理をdependsとして定義することで、コードが整理され、存在しないIDの場合のエラーハンドリングも一箇所にまとめられます。
“`python
from fastapi import Depends, FastAPI, HTTPException, Path, status
app = FastAPI()
ダミーのデータストア (実際はDBなど)
fake_items_db = {
“foo”: {“name”: “Foo”, “price”: 50.2},
“bar”: {“name”: “Bar”, “price”: 62.0},
“baz”: {“name”: “Baz”, “price”: 50.2, “disabled”: True},
}
item_id を基にアイテム情報を取得する依存関係
def get_item_from_db(item_id: str = Path(…, min_length=3)):
“””
item_id パスパラメータを基に、ダミーDBからアイテムを取得する
“””
# item_id は Path(…, min_length=3) で最低3文字であることを保証
print(f”Fetching item with id: {item_id}”)
if item_id not in fake_items_db:
# アイテムが見つからない場合は 404 エラー
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f”Item ‘{item_id}’ not found”
)
item = fake_items_db[item_id]
print(f”Found item: {item}”)
return item # 取得したアイテムオブジェクト (ここでは辞書) を返す
@app.get(“/items/{item_id}”)
def read_single_item(item: dict = Depends(get_item_from_db)):
“””
特定のアイテム情報を取得するエンドポイント
“””
# item は get_item_from_db から取得されたアイテム辞書
print(“Accessing item data within endpoint.”)
return item
例：
GET /items/foo -> Fetching item with id: foo -> Found item: {…} -> Accessing item data… -> {…} レスポンス
GET /items/baz -> Fetching item with id: baz -> Found item: {…} -> Accessing item data… -> {…} レスポンス
GET /items/nonexistent -> Fetching item with id: nonexistent -> HTTPException -> 404 Not Found
GET /items/ab -> Path() の min_length バリデーションに失敗 -> 422 Unprocessable Entity
“`
get_item_from_db依存関係は、パスパラメータitem_idを受け取り、ダミーのデータベースから対応するアイテムを検索します。アイテムが見つからない場合は、HTTPException(404)を発生させます。エンドポイント関数read_single_itemは、単にitem: dict = Depends(get_item_from_db)と宣言するだけで、存在が保証され、かつ取得済みのアイテムオブジェクト（辞書）を引数として受け取ることができます。
このパターンは、エンドポイント関数のコードを非常にシンプルにし、エラーハンドリングロジックを依存関係に集中させることができます。
依存関係のキャッシュ (use_cache=True)
デフォルトでは、同じリクエストのスコープ内で同じ依存関係が複数回使用された場合、FastAPIはその依存関係関数を一度だけ実行し、その結果をキャッシュして再利用します。これは、データベースセッションの取得や設定値の読み込みなど、リクエストごとに何度も実行する必要のない、コストのかかる依存関係にとって非常に効率的です。
依存関係関数をDepends()で指定する際に、明示的にキャッシュの挙動を制御することも可能です。Depends(your_dependency, use_cache=True)（デフォルト）またはDepends(your_dependency, use_cache=False)と指定します。
“`python
from fastapi import Depends, FastAPI
app = FastAPI()
キャッシュされる依存関係 (デフォルト)
def cached_dependency():
print(“Cached dependency called.”)
return “cached_value”
キャッシュされない依存関係
def non_cached_dependency():
print(“Non-cached dependency called.”)
return “non_cached_value”
@app.get(“/cache_test/”)
def test_cache(
# デフォルトでキャッシュされる
cached_val_1: str = Depends(cached_dependency),
# 明示的にキャッシュされる (デフォルトと同じ)
cached_val_2: str = Depends(cached_dependency, use_cache=True),
# キャッシュされない
non_cached_val_1: str = Depends(non_cached_dependency, use_cache=False),
non_cached_val_2: str = Depends(non_cached_dependency, use_cache=False),
):
“””
依存関係のキャッシュの挙動を確認するエンドポイント
“””
print(“Endpoint handler called.”)
return {
“cached_val_1”: cached_val_1,
“cached_val_2”: cached_val_2,
“non_cached_val_1”: non_cached_val_1,
“non_cached_val_2”: non_cached_val_2,
}
実行例：
GET http://localhost:8000/cache_test/
出力：
Cached dependency called. # 同じリクエスト内で一度だけ呼ばれる
Non-cached dependency called. # Depends(…, use_cache=False) なので複数回呼ばれる
Non-cached dependency called. # Depends(…, use_cache=False) なので複数回呼ばれる
Endpoint handler called.
レスポンス：
{“cached_val_1″:”cached_value”,”cached_val_2″:”cached_value”,”non_cached_val_1″:”non_cached_value”,”non_cached_val_2″:”non_cached_value”}
“`
この例からわかるように、Depends(cached_dependency)は同じリクエスト内で何度呼ばれても依存関係関数は一度しか実行されません。一方、Depends(non_cached_dependency, use_cache=False)は呼ばれるたびに依存関係関数が実行されます。
ほとんどの場合、デフォルトのuse_cache=Trueが望ましい動作です。特に、リソースの取得や初期化を行う依存関係はキャッシュされるべきです。キャッシュを無効にするのは、依存関係がリクエストの実行中に状態を変更するような、副作用を持つ場合や、意図的に毎回新しいインスタンスが必要な場合などに限られます。データベースセッションのようにyieldを使う依存関係も、デフォルトでキャッシュされます。yieldの前処理は一度だけ実行され、yieldの返り値がキャッシュされ、後処理はリクエスト完了時に一度だけ実行されます。
dependsの応用と高度なテクニック
dependsは単に関数だけでなく、クラス、さらに依存関係が別の依存関係に依存するという形で、より複雑なシナリオにも適用できます。
クラスベースの依存関係
依存関係として単なる関数ではなく、クラスを使用することも可能です。これは、依存関係が何らかの状態を持つ必要がある場合や、設定値をインスタンス変数として保持したい場合などに役立ちます。
クラスを依存関係として使うには、そのクラスが呼び出し可能（callable）である必要があります。これは、クラスに__call__メソッドを実装するか、または単にFastAPIがクラスそのものをインスタンス化して使用します。Depends()にクラスを渡すと、FastAPIはそのクラスをインスタンス化し、そのインスタンスをエンドポイント関数に注入します。
“`python
from fastapi import Depends, FastAPI, Query
app = FastAPI()
設定値を保持するダミークラス (実際は get_settings() などから注入)
class AppSettings:
def init(self):
# 通常はここで設定ファイルや環境変数から値を読み込む
# ここでは固定値
print(“AppSettings instance created.”)
self.default_limit = 10
self.max_limit = 100
ページネーションのパラメータを処理するクラスベースの依存関係
class PaginationParams:
def init(self, settings: AppSettings = Depends(AppSettings)):
# コンストラクタで別の依存関係 (AppSettings) を注入することも可能
self.settings = settings
print(“PaginationParams instance created.”)
# クラスインスタンス自身が呼び出し可能になる
def __call__(self,
skip: int = 0,
limit: int = Query(None)): # デフォルトは None にしておき、設定値を使用
print("PaginationParams __call__ method called.")
# 設定値を使ってデフォルト値や最大値を決定
effective_limit = limit if limit is not None else self.settings.default_limit
effective_limit = min(effective_limit, self.settings.max_limit) # 最大値を制限
return {"skip": skip, "limit": effective_limit}
@app.get(“/items/”)
def read_items_paginated(
# Depends() にクラスインスタンス (PaginationParams()) を渡す
# FastAPI は PaginationParams() をインスタンス化し、その call メソッドを呼び出す
# call メソッドの返り値が injection_params に注入される
injection_params: dict = Depends(PaginationParams()) # Depends にインスタンスを渡す！
):
“””
ページネーションパラメータを使用するエンドポイント
“””
print(“Endpoint handler called.”)
# injection_params は call メソッドの返り値
skip = injection_params[“skip”]
limit = injection_params[“limit”]
return {“message”: “Items paginated”, “skip”: skip, “limit”: limit}
実行例：
GET http://localhost:8000/items/
-> AppSettings instance created. (初回のみ)
PaginationParams instance created. (初回のみ)
PaginationParams call method called. (リクエストごと)
Endpoint handler called.
{“message”: “Items paginated”, “skip”: 0, “limit”: 10} # limit は AppSettings.default_limit
GET http://localhost:8000/items/?skip=20&limit=50
-> PaginationParams call method called.
Endpoint handler called.
{“message”: “Items paginated”, “skip”: 20, “limit”: 50}
GET http://localhost:8000/items/?skip=20&limit=200
-> PaginationParams call method called.
Endpoint handler called.
{“message”: “Items paginated”, “skip”: 20, “limit”: 100} # limit は AppSettings.max_limit に制限
“`
この例では、PaginationParamsというクラスを定義し、その__call__メソッドでページネーションに必要なskipとlimitパラメータを処理しています。__call__メソッドは、関数ベースの依存関係と同様に、リクエストパラメータ（skip, limitクエリパラメータ）を引数として受け取ることができます。
PaginationParamsクラスのコンストラクタは、別の依存関係であるAppSettingsに依存しています。FastAPIはDepends(PaginationParams())を解決する際に、まずPaginationParamsクラスをインスタンス化しようとします。その際にコンストラクタが必要とするAppSettingsを解決するために、Depends(AppSettings)を実行し、AppSettingsのインスタンスを生成してPaginationParamsコンストラクタに渡します。このように、依存関係は他の依存関係に依存することができます。
Depends(PaginationParams())と記述することで、FastAPIはPaginationParamsクラスのインスタンスを作成し、そのインスタンスを呼び出し可能オブジェクト（Callable）として扱います。クラスインスタンス自身が呼び出し可能であれば（__call__メソッドがあれば）、その__call__メソッドが実行され、その返り値が依存関係の最終的な値として注入されます。もし__call__メソッドがなければ、クラスインスタンスそのものが注入されます。
クラスベースの依存関係は、関連するメソッドや状態をまとめて管理したい場合に便利です。
サブ依存関係（Dependable）
依存関係は、他の依存関係に依存することができます。これは、より小さく、再利用可能な依存関係を組み合わせて、複雑な依存関係を構築する場合に役立ちます。FastAPIのドキュメントでは、依存関係として使用できるあらゆるものを「Dependable」と呼びます。関数、クラス、さらにはDepends()でラップされた別のDependableなどがこれにあたります。
前の例のPaginationParamsクラスのコンストラクタがAppSettingsに依存していたように、関数ベースの依存関係も他の依存関係に依存できます。
“`python
from fastapi import Depends, FastAPI, Query
app = FastAPI()
サブ依存関係 1：ユーザーIDを取得し、基本的な認証を行う (ダミー)
def get_current_user_id(user_id_header: str = Query(…, alias=”X-User-ID”)):
“””
X-User-ID クエリパラメータからユーザーIDを取得する (ダミー認証)
“””
# 実際はトークン検証などを行う
if user_id_header != “valid_user_id”:
raise HTTPException(status_code=401, detail=”Authentication failed”)
print(f”Authenticated user_id: {user_id_header}”)
return user_id_header
サブ依存関係 2：ユーザーIDを基にユーザーオブジェクトを取得する (ダミー)
def get_user_object(user_id: str = Depends(get_current_user_id)):
“””
ユーザーID (get_current_user_id の返り値) を基にユーザーオブジェクトを取得
“””
print(f”Fetching user object for user_id: {user_id}”)
# 実際はデータベースからユーザー情報を取得
if user_id == “valid_user_id”:
return {“id”: user_id, “name”: “Authenticated User”, “role”: “user”}
# get_current_user_id が既に認証済みなので、この部分の実行は理論上ないが、念のため
raise HTTPException(status_code=404, detail=”User not found”)
メインの依存関係：認証済みユーザーオブジェクトと権限チェックを行う
def get_current_active_user(user: dict = Depends(get_user_object)):
“””
認証済みユーザーオブジェクトを取得し、アクティブであるかチェック (ダミー)
“””
print(f”Checking if user ‘{user[‘id’]}’ is active.”)
# 実際はユーザーオブジェクトの属性などをチェック
# if user.is_active == False:
# raise HTTPException(status_code=400, detail=”Inactive user”)
return user
管理者ユーザーであることを確認する依存関係
def get_current_admin_user(user: dict = Depends(get_current_active_user)):
“””
認証済みアクティブユーザーを取得し、管理者権限があるかチェック (ダミー)
“””
print(f”Checking if user ‘{user[‘id’]}’ is admin.”)
if user.get(“role”) != “admin”: # ダミーで role をチェック
raise HTTPException(status_code=403, detail=”Admin privilege required”)
return user
app = FastAPI()
認証済みユーザーのみアクセス可能
@app.get(“/users/me/”)
def read_users_me(current_user: dict = Depends(get_current_active_user)):
“””
現在の認証済みユーザー情報を取得
“””
# current_user は get_current_active_user の返り値
print(“Endpoint handler called for /users/me/.”)
return current_user
管理者ユーザーのみアクセス可能
@app.get(“/admin/”)
def read_admin_info(current_admin: dict = Depends(get_current_admin_user)):
“””
管理者情報にアクセス
“””
# current_admin は get_current_admin_user の返り値
print(“Endpoint handler called for /admin/.”)
return {“message”: “Welcome, admin!”, “admin_info”: current_admin}
例：
GET /users/me/?X-User-ID=valid_user_id
-> Authenticated user_id: valid_user_id
Fetching user object for user_id: valid_user_id
Checking if user ‘valid_user_id’ is active.
Endpoint handler called for /users/me/.
{“id”:”valid_user_id”,”name”:”Authenticated User”,”role”:”user”} レスポンス
GET /admin/?X-User-ID=valid_user_id
-> Authenticated user_id: valid_user_id
Fetching user object for user_id: valid_user_id
Checking if user ‘valid_user_id’ is active.
Checking if user ‘valid_user_id’ is admin.
HTTPException -> 403 Forbidden
“`
この例では、
get_current_user_idがリクエストからユーザーIDを取得・認証します。get_user_objectが、get_current_user_idの返り値（ユーザーID）に依存して、ユーザーオブジェクトを取得します。get_current_active_userが、get_user_objectの返り値（ユーザーオブジェクト）に依存して、アクティブユーザーであることをチェックします。get_current_admin_userが、get_current_active_userの返り値（アクティブユーザーオブジェクト）に依存して、管理者権限をチェックします。
エンドポイント/users/me/はget_current_active_userに依存し、/admin/はget_current_admin_userに依存しています。FastAPIは依存関係ツリーを自動的に解決し、下位の依存関係から順に実行していきます。この構造により、認証 -> ユーザー取得 -> アクティブチェック -> 管理者チェック、という段階的な処理を、それぞれ独立した再利用可能な関数として定義できます。
グローバルな依存関係とルーターごとの依存関係
特定のAPIパスプレフィックスを持つすべてのエンドポイントや、アプリケーション全体のすべて・ほとんどのエンドポイントに対して、共通の依存関係（例：認証、DBセッション）を適用したい場合があります。これを実現するために、APIRouterやFastAPIインスタンスのdependenciesパラメータを使用できます。
“`python
from fastapi import Depends, FastAPI, HTTPException, APIRouter
共通の依存関係 (例: 基本的な認証)
def common_auth(auth_header: str = Header(…, alias=”Authorization”)):
“””
Authorization ヘッダーをチェックする共通認証 (ダミー)
“””
print(f”Checking Authorization header: {auth_header}”)
# 実際はトークン検証など
if auth_header != “Bearer common_token”:
raise HTTPException(status_code=401, detail=”Common auth failed”)
return True # 認証成功
管理者向けの認証依存関係 (より強い認証や権限チェック)
def admin_auth(admin_header: str = Header(…, alias=”X-Admin-Secret”)):
“””
管理者向けの特別なヘッダーをチェック (ダミー)
“””
print(f”Checking X-Admin-Secret header: {admin_header}”)
if admin_header != “super_secret”:
raise HTTPException(status_code=403, detail=”Admin access denied”)
return True # 認証成功
ルーター定義
items_router = APIRouter(
prefix=”/items”,
tags=[“items”],
# このルーター内のすべてのエンドポイントに common_auth を適用
dependencies=[Depends(common_auth)]
)
admin_router = APIRouter(
prefix=”/admin”,
tags=[“admin”],
# このルーター内のすべてのエンドポイントに admin_auth を適用
dependencies=[Depends(admin_auth)]
)
@items_router.get(“/”)
def read_items():
“””アイテム一覧を取得 (common_auth が適用される)”””
print(“/items/ endpoint called.”)
return [{“item_id”: “Foo”}, {“item_id”: “Bar”}]
@items_router.get(“/{item_id}”)
def read_item(item_id: str):
“””特定のアイテムを取得 (common_auth が適用される)”””
print(f”/items/{item_id} endpoint called.”)
return {“item_id”: item_id}
@admin_router.get(“/dashboard/”)
def read_admin_dashboard():
“””管理者ダッシュボード (admin_auth が適用される)”””
print(“/admin/dashboard/ endpoint called.”)
return {“message”: “Admin dashboard data”}
FastAPI アプリケーションインスタンス
app = FastAPI(
# アプリケーション全体のすべてのエンドポイントに別の共通依存関係を適用することも可能
# 例えば、DBセッションなどをここで適用することが多い
# dependencies=[Depends(get_db)] # get_db() は yield を使う依存関係の例
)
ルーターをアプリケーションに追加
app.include_router(items_router)
app.include_router(admin_router)
例：
GET /items/ -H “Authorization: Bearer common_token” -> common_auth OK -> /items/ endpoint -> 成功
GET /items/ -H “Authorization: Bearer wrong_token” -> common_auth failed -> 401 Unauthorized
GET /admin/dashboard/ -H “X-Admin-Secret: super_secret” -> admin_auth OK -> /admin/dashboard/ endpoint -> 成功
GET /admin/dashboard/ -H “X-Admin-Secret: wrong_secret” -> admin_auth failed -> 403 Forbidden
GET /admin/dashboard/ -H “Authorization: Bearer common_token” -> admin_auth required, Authorization ヘッダーは admin_auth では見ない -> 403 Forbidden (または X-Admin-Secret がないことによるバリデーションエラー)
“`
APIRouterのdependencies引数に依存関係のリストを指定すると、そのルーターに登録されたすべてのエンドポイント関数にその依存関係が自動的に適用されます。同様に、FastAPIインスタンスのdependencies引数に指定すると、アプリケーション全体に登録されたすべてのエンドポイントに適用されます。
これは、例えばDBセッションをアプリケーション全体で共有したい場合に非常に便利です。FastAPI(dependencies=[Depends(get_db)])のように設定すれば、すべてのエンドポイント関数は自動的にget_db依存関係を実行し、yieldされたセッションを受け取ることができます（ただし、エンドポイント関数でそのセッションを引数として受け取る必要はあります）。認証や権限チェックのように、パスプレフィックスによって適用したい依存関係を変えたい場合は、ルーターごとのdependenciesが適しています。
個別のエンドポイント関数で指定されたDependsは、ルーターやアプリケーション全体で指定されたdependenciesリストに追加して実行されます。したがって、共通認証をルーターで適用しつつ、特定の管理者エンドポイントではさらに管理者権限チェックの依存関係を追加するといった組み合わせも可能です。
テスト容易性
dependsを活用することで、APIエンドポイント関数のテストが劇的に容易になります。エンドポイント関数は、自身が依存する外部リソース（データベース接続、認証済みのユーザー情報、設定オブジェクトなど）を引数として受け取るように設計されています。これにより、テスト時にはこれらの依存関係を実際のオブジェクトの代わりに「モック」（ダミーの代替オブジェクト）に差し替えることができます。
FastAPIは、この目的のためにapp.dependency_overridesという機能を提供しています。これは、テスト中に特定の依存関係が解決される際に、本来の依存関係関数の代わりに別の関数（モック）を実行するようにFastAPIに指示するものです。
“`python
main.py (FastAPI アプリケーションコード)
from fastapi import Depends, FastAPI, HTTPException
def get_user_from_db(user_id: int):
“””データベースからユーザーを取得する (ダミー)”””
print(f”Fetching user {user_id} from real DB.”)
if user_id == 1:
return {“id”: 1, “name”: “Real User”}
return None
def get_current_active_user_from_db(user_id: int = Depends(get_user_from_db)):
“””アクティブユーザーを取得 (ダミー)”””
if user_id is None:
raise HTTPException(status_code=404, detail=”User not found”)
print(f”Checking active status for user {user_id}.”)
# 実際のアクティブチェックロジック
return user_id # ユーザーが見つかったら ID を返す
app = FastAPI()
@app.get(“/users/{user_id}”)
def read_user(
user_id: int, # パスパラメータはそのまま受け取る
active_user_id: int = Depends(get_current_active_user_from_db) # 依存関係の結果を受け取る
):
“””
データベースからユーザーを取得し、アクティブかチェックするエンドポイント
“””
print(“Endpoint handler called.”)
# active_user_id は get_current_active_user_from_db の返り値
# ここでは active_user_id を使って何か処理を行う
return {“user_id”: user_id, “active_user_id_from_dependency”: active_user_id, “message”: “User is active”}
test_main.py (テストコード)
from fastapi.testclient import TestClient
main.py から app をインポート
from .main import app, get_user_from_db, get_current_active_user_from_db
FastAPI アプリケーションインスタンス
(テスト実行時にはテストクライアントを使用)
app = FastAPI() # 実際は main.py からインポート
def get_user_from_db(user_id: int): … # 実際は main.py からインポート
def get_current_active_user_from_db(user_id: int = Depends(get_user_from_db)): … # 実際は main.py からインポート
@app.get(“/users/{user_id}”) def read_user(…): … # 実際は main.py からインポート
client = TestClient(app)
モックとなる依存関係関数を定義
def override_get_user_from_db(user_id: int):
“””データベースアクセスをモックする関数”””
print(f”Fetching user {user_id} from MOCK DB.”)
if user_id == 1:
return {“id”: 1, “name”: “Mock User”} # テスト用のダミーユーザー
return None
def override_get_current_active_user_from_db(user_id: int = Depends(override_get_user_from_db)):
“””アクティブユーザーチェックをモックする関数”””
print(f”Checking active status for user {user_id} using MOCK.”)
if user_id is None:
# HTTPException をモック内で発生させることも可能
raise HTTPException(status_code=404, detail=”User not found (mock)”)
# モックとして常にアクティブとみなすなど
return user_id
def test_read_user_success():
# dependency_overrides を設定
# key: 本来の依存関係関数, value: モック関数
app.dependency_overrides[get_user_from_db] = override_get_user_from_db
app.dependency_overrides[get_current_active_user_from_db] = override_get_current_active_user_from_db
# テスト実行
response = client.get("/users/1")
# dependency_overrides を解除 (テスト終了後)
app.dependency_overrides = {}
assert response.status_code == 200
assert response.json() == {"user_id": 1, "active_user_id_from_dependency": 1, "message": "User is active"}
def test_read_user_not_found():
# dependency_overrides を設定
app.dependency_overrides[get_user_from_db] = override_get_user_from_db
app.dependency_overrides[get_current_active_user_from_db] = override_get_current_active_user_from_db
# 存在しないユーザーIDでテスト実行
response = client.get("/users/99")
# dependency_overrides を解除
app.dependency_overrides = {}
assert response.status_code == 404
assert response.json() == {"detail": "User not found (mock)"}
実行例 (pytest などで実行した場合):
Fetching user 1 from MOCK DB.
Checking active status for user 1 using MOCK.
Endpoint handler called.
.
Fetching user 99 from MOCK DB.
Checking active status for user 99 using MOCK.
.
2 passed in …s
“`
app.dependency_overridesは辞書です。キーとしてオーバーライドしたい元の依存関係関数（またはクラスなど）、値としてその代わりに実行したいモック関数（またはクラス、値など）を指定します。テストケースの実行前にこれを設定し、テスト後に忘れずにクリア（またはテストフィクスチャで管理）することが重要です。
この機能を使うことで、データベース、外部API、認証システムなど、外部への依存を持つエンドポイント関数を、これらの依存先と実際に通信することなくテストできるようになります。これにより、テストが高速化され、テスト環境のセットアップが簡略化され、単体テストの範囲を正確に定めることができます。
dependsを使う上での注意点とベストプラクティス
dependsは非常に強力な機能ですが、適切に使用しないとコードが読みにくくなったり、予期せぬ挙動を引き起こしたりする可能性もあります。
- 依存関係の粒度: 依存関係関数は、単一の明確な責務を持つように小さく保つことが望ましいです。例えば、「認証と同時にデータベースからユーザー情報を取得し、権限チェックも行う」という大きな依存関係を作るのではなく、「トークンを検証してユーザーIDを取得する」「ユーザーIDを基にユーザーオブジェクトを取得する」「ユーザーオブジェクトを基に権限をチェックする」といった、より小さな依存関係に分割し、それらを組み合わせて使用する方が、再利用性が高く、テストもしやすくなります（サブ依存関係の例を参照）。
- エラーハンドリング: 依存関係内でエラーが発生した場合は、
fastapi.HTTPExceptionを送出するのが標準的な方法です。これにより、FastAPIは適切なHTTPステータスコードとレスポンスボディでクライアントに応答します。依存関係内で予期しない例外が発生した場合、FastAPIはデフォルトで500 Internal Server Errorとして扱いますが、これは一般的な挙動であり問題ありません。 yieldを使った依存関係の終了処理:yieldを使った依存関係（Generator Dependencies）で、yieldの後に記述するクリーンアップコード（例：DBセッションのクローズ）は、エンドポイント関数が正常終了した場合だけでなく、例外が発生した場合でも確実に実行されるように設計することが重要です。try...except...finallyブロックを適切に使用してください。リソースリークを防ぐ上でこれは不可欠です。- 循環依存関係の回避: 依存関係AがBに依存し、BがAに依存する、といった循環依存関係は FastPI が解決できず、エラーになります。依存関係の構造を設計する際には、依存関係の方向が一方向になるように注意が必要です。サブ依存関係を構築する際には、依存ツリーが循環しないかを確認しましょう。
- ドキュメンテーション（OpenAPIスキーマ）:
fastapi.securityモジュールのクラス（APIKeyHeader,OAuth2PasswordBearerなど）や、fastapi.param_utils.Depends（実際はfastapi.params.Dependsなどを使用）を適切に使用することで、依存関係が要求するパラメータ（ヘッダー、クエリなど）やセキュリティ情報が自動生成されるOpenAPIスキーマに正確に反映されます。これはAPIドキュメントの質を高める上で重要です。カスタムの依存関係で特別なパラメータを要求する場合は、Query,Path,Header,Cookieなどのパラメータ定義オブジェクトを使用することでドキュメントに反映されます。 - 非同期処理: 依存関係関数も非同期関数（
async def）として定義できます。非同期の依存関係は、データベースアクセスや外部API呼び出しなど、I/Oバウンドな処理を行う場合に特に効果的です。FastAPIは非同期依存関係も適切に解決します。 - 副作用とキャッシュ: デフォルトの
use_cache=Trueの挙動は効率的ですが、もし依存関係関数がリクエストの状態を変更するなどの副作用を持つ場合は、use_cache=Falseを検討する必要があるかもしれません。ただし、副作用を持つ依存関係は一般的に避けるべき設計パターンです。状態が必要な場合は、クラスベースの依存関係や、アプリケーションの状態を適切に管理する方が望ましいです。
まとめ
FastAPIのdependsは、依存関係注入（DI）の原則をAPIエンドポイントに適用するための、フレームワークの中核をなす機能です。この記事では、dependsの基本的な使い方から、認証・認可、データベースセッション管理、設定値の読み込みといった共通処理の効率的な実装方法、さらにはクラスベースの依存関係、サブ依存関係、グローバルな依存関係、そしてテスト時のモックといった高度な活用方法までを詳細に見てきました。
dependsを使いこなすことで、
- コードの再利用性が向上します。 共通のロジックを依存関係として抽出し、複数のエンドポイントで使い回せます。
- コードがクリーンになります。 エンドポイント関数は自身の主要な責務に集中でき、周辺の共通処理は依存関係に委ねられます。
- テストが容易になります。 依存関係をモックに差し替えることで、外部依存なしにエンドポイント関数のユニットテストを書けます。
- 保守性が高まります。 共通処理の変更が必要になった場合、修正は依存関係関数一箇所で行えば済みます。
- APIの構造が明確になります。 エンドポイント関数のシグネチャを見るだけで、そのAPIがどのような依存関係（認証が必要か、特定のパラメータが必要か、DBセッションが必要かなど）を持つかが分かりやすくなります。
FastAPIのdependsは、単なる関数の呼び出し以上のものです。それは、FastAPIアプリケーション全体の構造、保守性、テスト容易性に大きな影響を与える設計パターンを促進する強力なツールです。この記事で解説した様々なテクニックを参考に、あなたのFastAPI開発においてdependsを最大限に活用し、より堅牢で効率的なAPIを構築してください。