FastAPIでのミドルウェア追加方法
FastAPIは、Webアプリケーションの各リクエスト/レスポンスサイクルに処理を挿入できるミドルウェアの概念をサポートしています。これは、認証、ロギング、 CORS (Cross-Origin Resource Sharing) の設定、リクエストの検証、レスポンスの加工など、アプリケーション全体に適用したい共通の機能を実装するのに非常に役立ちます。
ミドルウェアの基本概念
ミドルウェアは、リクエストが最終的なエンドポイント（ルーター関数）に到達する前、またはレスポンスがクライアントに送信される前に実行される関数です。FastAPIでは、ASGI (Asynchronous Server Gateway Interface) アプリケーションの機能を利用してミドルウェアを実装します。
ASGIミドルウェアは、以下のシグネチャを持つ呼び出し可能なオブジェクト（関数またはクラスのインスタンス）として定義されます。
async def middleware_function(request: Request, call_next):
# リクエストに対する処理
response = await call_next(request)
# レスポンスに対する処理
return response
ここで、
request: クライアントからのHTTPリクエストを表すStarletteのRequestオブジェクトです。call_next: 次のミドルウェアまたは最終的なエンドポイントにリクエストを渡すための非同期関数です。この関数を呼び出すことで、リクエスト処理の連鎖が進みます。response:call_nextから返されるHTTPレスポンスを表すStarletteのResponseオブジェクトです。
FastAPIアプリケーションは、ASGIアプリケーションであるため、このASGIミドルウェアの概念をそのまま適用できます。
FastAPIでのミドルウェアの追加方法
FastAPIアプリケーションにミドルウェアを追加するには、主に2つの方法があります。
1. ASGIアプリケーションに直接追加
FastAPIインスタンス（FastAPI()）は、ASGIアプリケーションです。このASGIアプリケーションのmiddleware属性に、ASGIミドルウェアをリストとして追加することができます。
コード例
from fastapi import FastAPI, Request
from starlette.middleware.cors import CORSMiddleware
app = FastAPI()
# ミドルウェア関数を定義
async def custom_middleware(request: Request, call_next):
print("リクエスト受信前処理")
response = await call_next(request)
print("レスポンス送信前処理")
return response
# FastAPIインスタンスにミドルウェアを追加
app.middleware("http")(custom_middleware)
# CORSミドルウェアも追加する場合
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def read_root():
return {"Hello": "World"}
この例では、custom_middlewareという自作のミドルウェア関数と、CORSMiddlewareという標準のミドルウェアを追加しています。
app.middleware("http")(custom_middleware): “http” プロトコル向けのミドルウェアとしてcustom_middlewareを登録します。app.add_middleware(...):add_middlewareメソッドは、より高レベルなインターフェースを提供し、CORSMiddlewareのようなクラスベースのミドルウェアを簡単に設定できます。
ミドルウェアは、追加された順序で実行されます。そのため、実行順序を考慮して追加することが重要です。
2. StarletteのMiddlewareクラスを使用
FastAPIはStarletteをベースにしているため、StarletteのMiddlewareクラスを使用してミドルウェアを定義し、それをFastAPIアプリケーションのmiddlewareリストに追加することもできます。
コード例
from fastapi import FastAPI, Request
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
async def custom_middleware_v2(request: Request, call_next):
print("カスタムミドルウェアv2: リクエスト受信")
response = await call_next(request)
print("カスタムミドルウェアv2: レスポンス送信")
return response
# Middlewareオブジェクトを作成
middleware_list = [
Middleware(CORSMiddleware, allow_origins=["*"]),
Middleware(custom_middleware_v2), # 関数を直接渡すことも可能
]
app = FastAPI(middleware=middleware_list)
@app.get("/")
async def read_root():
return {"Hello": "World v2"}
この方法では、FastAPIアプリケーションの初期化時にmiddleware引数にMiddlewareオブジェクトのリストを渡すことで、ミドルウェアを設定します。
ミドルウェアのユースケース例
FastAPIでのミドルウェアは、様々なシナリオで活用できます。
例1: リクエストロギング
from fastapi import FastAPI, Request
import datetime
app = FastAPI()
async def logging_middleware(request: Request, call_next):
start_time = datetime.datetime.now()
print(f"[{start_time.isoformat()}] {request.method} {request.url}")
response = await call_next(request)
end_time = datetime.datetime.now()
process_time = (end_time - start_time).total_seconds()
print(f"[{end_time.isoformat()}] {request.method} {request.url} {response.status_code} {process_time:.2f}s")
return response
app.middleware("http")(logging_middleware)
@app.get("/")
async def read_root():
return {"message": "Logged"}
このミドルウェアは、リクエストの受信時とレスポンスの送信時にタイムスタンプ、HTTPメソッド、URL、ステータスコード、処理時間をコンソールに出力します。
例2: カスタム認証
from fastapi import FastAPI, Request, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response
class CustomAuthMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
# ここで認証ロジックを実装
# 例: Authorizationヘッダーのチェック
auth_header = request.headers.get("Authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return Response("Unauthorized", status_code=401)
# 認証成功後、リクエストを次に渡す
response = await call_next(request)
return response
app = FastAPI()
app.add_middleware(CustomAuthMiddleware)
@app.get("/protected")
async def protected_route():
return {"message": "Access granted"}
@app.get("/")
async def public_route():
return {"message": "Public"}
BaseHTTPMiddlewareを継承することで、より構造化されたミドルウェアを実装できます。この例では、Authorizationヘッダーの存在をチェックし、存在しない場合は401 Unauthorizedを返します。
例3: リクエストボディの加工
from fastapi import FastAPI, Request
import json
app = FastAPI()
async def process_request_body(request: Request, call_next):
if request.method == "POST" and request.headers.get("content-type") == "application/json":
try:
body_bytes = await request.body()
body_dict = json.loads(body_bytes)
# リクエストボディを加工する例
body_dict["processed"] = True
body_bytes_processed = json.dumps(body_dict).encode("utf-8")
# リクエストオブジェクトを再構築（注意：これは通常、直接は推奨されません。Starletteの機能を使います）
# より洗練された方法は、bodyを読み取った後に、bodyをrequestに再設定するのではなく、
# 新しいリクエストオブジェクトを作成して渡すか、
# content-typeなどを変更しないように注意が必要です。
# ここでは簡略化のために、bodyを読み替えてnextに渡します。
# 実際には、request.scope["body"]などを操作する場合がありますが、
# これは内部実装に依存し、注意が必要です。
# より安全な方法は、リクエストボディを読み取った後、
# call_nextに渡す際に、加工したデータをbodyとして渡すためのラッパーなどを検討することです。
# FastAPI/Starletteのrequestオブジェクトはimmutableな部分があるため、
# bodyを直接変更するのは複雑になることがあります。
# 簡略化のため、ここではbodyを読み替えるという概念を示します。
# 実際には、bodyを読み取った後、
# request.scope["body"] = body_bytes_processed のような操作が考えられますが、
#これは非推奨となる可能性があります。
# より一般的なのは、リクエストボディを読み取った後、
# call_nextに渡す前に、リクエストの`state`属性などに加工したデータを格納し、
# ルート関数でそれを取り出す方法です。
# この例では、bodyを読み取った後、request.scope["raw_body"]などに格納して、
# call_nextでbodyを読み取る際に、その加工されたbodyを読むようにする、
# というような複雑な実装が必要になります。
# 簡略化のため、ここではbodyを読み取って加工し、
# その加工したbodyを次の処理に渡すという「概念」に留めます。
# 実際には、request.scope['body'] を直接変更することは、
# Starletteの内部実装に依存するため、一般的には推奨されません。
# より安全なアプローチは、request.scope['body'] を読み取った後、
# body_bytes_processed のような加工済みのバイト列を、
# call_next に渡されるリクエストオブジェクトのbodyとして利用できるようにすることです。
# しかし、FastAPI/StarletteのRequestオブジェクトのbodyは、
# await request.body() を呼び出すと消費されてしまうため、
# 再度利用するには、request.scope['body'] などを操作する必要があります。
# ここでは、read_body() を呼び出した後、
# request.scope["_body"] に加工したbodyを格納し、
# call_next で bodyを読み込む際に、この "_body" を優先的に読み込むような、
# カスタムのRequestラッパーを定義するなどの方法が考えられます。
# 簡略化のために、ここではbodyを加工したことを示すprint文に留めます。
print("リクエストボディを加工しました。")
except Exception as e:
print(f"リクエストボディの処理中にエラー: {e}")
# エラーが発生した場合、エラーレスポンスを返す
return Response("Bad Request", status_code=400)
response = await call_next(request)
return response
app.middleware("http")(process_request_body)
@app.post("/items/")
async def create_item(request: Request):
# ここで、ミドルウェアで加工されたボディを利用できるようにする
# 実際には、ミドルウェアでbodyを加工し、その結果をルート関数で利用するための
# mechanism を別途用意する必要があります。
# 例えば、ミドルウェアで加工したデータを request.state.processed_data のような形で格納し、
# ルート関数で request.state.processed_data を参照するなど。
# await request.json() はミドルウェアでawait request.body() を実行しているため、
# 再度実行するとエラーになる可能性があります。
# そのため、ミドルウェアで body を読み取った後、
# request.state に加工したデータを設定するのが一般的です。
# body_dict = await request.json() # ミドルウェアで既にbodyを消費しているため、これはエラーになる可能性が高い
# return {"received": body_dict}
# ミドルウェアでbodyを加工したという事実のみを返します。
# 実際には、request.stateなどから加工されたデータを受け取る必要があります。
return {"message": "Item processed (concept)"}
この例は、リクエストボディを読み取り、JSONとしてパースし、加工する試みを示しています。しかし、FastAPI/StarletteのRequestオブジェクトのbodyは一度読み取ると消費されてしまうため、ミドルウェアでawait request.body()を実行した場合、後続のルーター関数では再度await request.body()やawait request.json()を直接実行できなくなります。この問題を回避するためには、ミドルウェアでリクエストボディを読み取った後、加工したデータをrequest.stateなどの属性に格納し、ルーター関数でそのrequest.stateからデータを取り出す、といった設計が必要になります。
ミドルウェアの注文と依存関係
ミドルウェアは、app.middleware()やapp.add_middleware()で追加された順序で実行されます。リクエストは、最初に追加されたミドルウェアから順に処理され、call_nextによって次のミドルウェアへ渡されます。最終的に、すべてのミドルウェアを通過したリクエストがルーター関数に到達します。レスポンスは、ルーター関数で生成された後、追加されたミドルウェアの逆順に処理されてクライアントに返されます。
この実行順序は非常に重要です。例えば、認証ミドルウェアをロギングミドルウェアよりも後に配置すると、認証に失敗したリクエストであってもロギングされてしまう可能性があります。逆に、認証ミドルウェアを先に配置することで、認証に失敗したリクエストは早期に処理を中断させることができます。
まとめ
FastAPIにおけるミドルウェアは、アプリケーションの共通機能を効果的に実装するための強力なメカニズムです。リクエスト/レスポンスのライフサイクルに処理を挿入することで、認証、ロギング、CORS設定、リクエスト・レスポンスの加工など、様々なタスクをクリーンかつ効率的に管理できます。app.middleware()やapp.add_middleware()、またはStarletteのMiddlewareクラスを使用して、柔軟にミドルウェアを追加・設定できます。ミドルウェアの実行順序を理解し、適切に設計することで、堅牢で保守性の高いWebアプリケーションを構築することが可能です。
