FastAPI ミドルウェアの作り方：認証、ロギング、エラー処理を実装

By /

FastAPI ミドルウェア詳解：認証、ロギング、エラー処理の実装

FastAPI は、モダンな Python Web API フレームワークとして、高速な開発速度、自動ドキュメント生成、そして何よりも柔軟性と拡張性に優れています。 その柔軟性を支える重要な要素の一つが、ミドルウェアです。 ミドルウェアは、リクエストとレスポンスの処理パイプラインに介入し、認証、ロギング、エラー処理など、アプリケーション全体に共通する機能を効率的に実装するための強力なツールとなります。

本稿では、FastAPI ミドルウェアの基本的な概念から、具体的な実装方法までを詳細に解説します。 認証、ロギング、エラー処理といった代表的なユースケースを例に、実用的なミドルウェアの構築方法をステップバイステップで学びましょう。

1. ミドルウェアとは何か？

ミドルウェアとは、アプリケーションのリクエストとレスポンスの間に位置するソフトウェアコンポーネントです。 リクエストがアプリケーションに到達する前、そしてレスポンスがクライアントに送信される前に、ミドルウェアは処理を行うことができます。 この特性を利用して、アプリケーションのコアロジックに影響を与えることなく、様々な機能を付加したり変更したりすることが可能です。

FastAPI におけるミドルウェアは、ASGI (Asynchronous Server Gateway Interface) を利用して実装されます。 ASGI は、非同期処理をサポートする Web サーバーとアプリケーション間の標準インターフェースであり、FastAPI の高速性と効率性を支える基盤技術です。

2. FastAPI におけるミドルウェアの定義と登録

FastAPI でミドルウェアを定義するには、関数またはクラスを使用します。 関数を使用する場合は、async 関数である必要があります。 クラスを使用する場合は、__call__ メソッドを実装する必要があります。

2.1 関数ベースのミドルウェア

関数ベースのミドルウェアは、最もシンプルで基本的なミドルウェアの実装方法です。 以下に示すように、async 関数を定義し、@app.middleware("http") デコレータを使用して FastAPI アプリケーションに登録します。

“`python
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse

app = FastAPI()

@app.middleware(“http”)
async def add_process_time_header(request: Request, call_next):
“””
リクエスト処理時間をヘッダーに追加するミドルウェア
“””
import time
start_time = time.time()
response = await call_next(request) # 次のミドルウェアまたはルーティング関数を呼び出す
process_time = time.time() – start_time
response.headers[“X-Process-Time”] = str(process_time)
return response

@app.get(“/”)
async def root():
return {“message”: “Hello World”}
“`

この例では、add_process_time_header 関数がミドルウェアとして定義されています。

  • request: Request は、受信した HTTP リクエストを表す Request オブジェクトです。 これを通じて、リクエストヘッダー、クッキー、ボディなどの情報にアクセスできます。
  • call_next は、次に実行されるミドルウェアまたはルーティング関数を呼び出すための非同期関数です。 ミドルウェアは、call_next を呼び出す前にリクエストを処理し、call_next の戻り値であるレスポンスを受け取った後にレスポンスを処理できます。

@app.middleware("http") デコレータは、この関数が HTTP リクエストのミドルウェアとして登録されることを FastAPI に伝えます。 FastAPI は、リクエストを受け取るたびに、登録されたミドルウェアを順番に実行します。

2.2 クラスベースのミドルウェア

クラスベースのミドルウェアは、より複雑なロジックや状態を保持する必要がある場合に適しています。 クラスは、__call__ メソッドを実装する必要があります。

“`python
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse
from typing import Callable

app = FastAPI()

class AuthenticationMiddleware:
def init(self, app):
self.app = app

async def __call__(self, request: Request, call_next: Callable):
    """
    認証処理を行うミドルウェア
    """
    # 例えば、リクエストヘッダーから認証トークンを取得
    authorization = request.headers.get("Authorization")
    if authorization == "Bearer mysecrettoken":
        # 認証成功
        response = await call_next(request)
        return response
    else:
        # 認証失敗
        return JSONResponse(status_code=401, content={"message": "Unauthorized"})

app.add_middleware(AuthenticationMiddleware)

@app.get(“/”)
async def root():
return {“message”: “Hello World”}
“`

この例では、AuthenticationMiddleware クラスがミドルウェアとして定義されています。

  • __init__ メソッドは、ミドルウェアの初期化処理を行います。 ここでは、FastAPI アプリケーションインスタンス (app) を保持しています。
  • __call__ メソッドは、ミドルウェアの主要な処理ロジックを実装します。 request と call_next の引数は、関数ベースのミドルウェアと同様です。

app.add_middleware(AuthenticationMiddleware) は、このクラスをミドルウェアとして FastAPI アプリケーションに登録します。

3. ミドルウェアの実行順序

複数のミドルウェアが登録されている場合、FastAPI はそれらを登録された順に実行します。 これは、ミドルウェアの処理順序が非常に重要であることを意味します。 例えば、ロギングミドルウェアの前に認証ミドルウェアを実行することで、認証に失敗したリクエストはロギングされないようにすることができます。

4. 代表的なミドルウェアの実装例

4.1 認証ミドルウェア

認証ミドルウェアは、API エンドポイントへのアクセスを制御するために不可欠です。 一般的な認証方式としては、API キー、OAuth 2.0、JWT (JSON Web Token) などがあります。

“`python
from fastapi import FastAPI, Request, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from starlette.responses import JSONResponse
from typing import Callable

app = FastAPI()
security = HTTPBearer()

async def verify_token(token: str):
“””
トークンの検証を行う関数
(ここでは単純な文字列比較を行っていますが、実際にはより複雑な処理が必要になります)
“””
# 実際にはデータベースや外部サービスでトークンを検証する
if token == “mysecrettoken”:
return True
else:
return False

@app.middleware(“http”)
async def authentication_middleware(request: Request, call_next: Callable):
“””
認証処理を行うミドルウェア
“””
try:
credentials = await security(request)
token = credentials.credentials
if await verify_token(token):
response = await call_next(request)
return response
else:
raise HTTPException(status_code=401, detail=”Invalid token”)
except HTTPException as e:
return JSONResponse(status_code=e.status_code, content={“message”: e.detail})
except Exception:
return JSONResponse(status_code=401, content={“message”: “Unauthorized”})
“`

この例では、authentication_middleware が認証ミドルウェアとして定義されています。

  • HTTPBearer は、HTTP Bearer 認証スキームを実装するための FastAPI ユーティリティクラスです。
  • security(request) は、リクエストヘッダーから認証情報を抽出します。
  • verify_token 関数は、トークンの検証を行う関数です。 (ここでは単純な文字列比較を行っていますが、実際にはデータベースや外部サービスでトークンを検証する必要があります。)
  • 認証に成功した場合は、call_next を呼び出して次のミドルウェアまたはルーティング関数を実行します。
  • 認証に失敗した場合は、401 Unauthorized エラーを返します。

4.2 ロギングミドルウェア

ロギングミドルウェアは、リクエストとレスポンスに関する情報を記録するために使用されます。 これにより、アプリケーションの動作を監視し、問題を診断し、パフォーマンスを分析することができます。

“`python
import logging
from fastapi import FastAPI, Request
from typing import Callable

app = FastAPI()

ロガーの設定

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(name)

@app.middleware(“http”)
async def logging_middleware(request: Request, call_next: Callable):
“””
リクエストとレスポンスに関する情報を記録するミドルウェア
“””
import time
start_time = time.time()

logger.info(f"Request: {request.method} {request.url}")
try:
    response = await call_next(request)
except Exception as e:
    logger.error(f"Exception: {e}")
    raise e
finally:
    process_time = time.time() - start_time
    logger.info(f"Response: {response.status_code} - Process Time: {process_time:.4f}s")
    return response

“`

この例では、logging_middleware がロギングミドルウェアとして定義されています。

  • logger.info は、リクエストとレスポンスに関する情報をログに記録します。
  • time.time() を使用して、リクエスト処理時間を計測します。
  • try...except...finally ブロックを使用して、例外が発生した場合でもレスポンスを返すようにします。

4.3 エラー処理ミドルウェア

エラー処理ミドルウェアは、アプリケーション内で発生した例外をキャッチし、適切なエラーレスポンスを返すために使用されます。 これにより、クライアントにわかりやすいエラーメッセージを提供し、アプリケーションの安定性を向上させることができます。

“`python
from fastapi import FastAPI, Request
from starlette.responses import JSONResponse
from typing import Callable

app = FastAPI()

@app.middleware(“http”)
async def error_handling_middleware(request: Request, call_next: Callable):
“””
アプリケーション内で発生した例外をキャッチし、適切なエラーレスポンスを返すミドルウェア
“””
try:
response = await call_next(request)
return response
except Exception as e:
# 例外の種類に応じて適切なエラーレスポンスを返す
if isinstance(e, ValueError):
status_code = 400
message = “Invalid input data”
elif isinstance(e, KeyError):
status_code = 404
message = “Resource not found”
else:
status_code = 500
message = “Internal server error”
return JSONResponse(status_code=status_code, content={“message”: message})
“`

この例では、error_handling_middleware がエラー処理ミドルウェアとして定義されています。

  • try...except ブロックを使用して、例外をキャッチします。
  • 例外の種類に応じて、適切なステータスコードとエラーメッセージを返します。

5. CORS (Cross-Origin Resource Sharing) ミドルウェア

CORS は、異なるオリジン (ドメイン、プロトコル、ポート) からの Web ページが互いにリソースを共有することを制限するセキュリティメカニズムです。 FastAPI には、CORS を制御するための CORSMiddleware が用意されています。

“`python
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

origins = [
“http://localhost”,
“http://localhost:8080”,
“https://example.com”,
]

app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=[““], # すべての HTTP メソッドを許可
allow_headers=[““], # すべてのヘッダーを許可
)

@app.get(“/”)
async def root():
return {“message”: “Hello World”}
“`

  • allow_origins は、リソースへのアクセスを許可するオリジンのリストです。 "*" を指定すると、すべてのオリジンからのアクセスが許可されますが、セキュリティ上の理由から、本番環境では具体的なオリジンを指定することを推奨します。
  • allow_credentials は、Cookie や Authorization ヘッダーなどの資格情報を許可するかどうかを指定します。
  • allow_methods は、許可する HTTP メソッドのリストです。
  • allow_headers は、許可する HTTP ヘッダーのリストです。

6. まとめ

FastAPI ミドルウェアは、認証、ロギング、エラー処理など、アプリケーション全体に共通する機能を効率的に実装するための強力なツールです。 本稿では、ミドルウェアの基本的な概念から、具体的な実装方法までを詳細に解説しました。 これらの知識を活用して、安全で堅牢な FastAPI アプリケーションを開発してください。

7. さらなる学習

この詳細な説明が、FastAPI ミドルウェアの理解と実装に役立つことを願っています。 実践的な演習を通して、より深くミドルウェアの知識を習得し、独自のミドルウェアを開発してみてください。

コメントする

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

上部へスクロール