カスタムリクエストクラスとAPIRouteクラス¶
場合によっては、RequestクラスとAPIRouteクラスで使用されるロジックをオーバーライドしたい場合があります。
特に、これはミドルウェアのロジックの良い代替手段となる場合があります。
例えば、アプリケーションによって処理される前にリクエストボディを読み取ったり操作したりしたい場合などです。
危険
これは「高度な」機能です。
FastAPIを始めたばかりの場合は、このセクションをスキップしてもよいでしょう。
ユースケース¶
いくつかのユースケースには以下が含まれます
- 非JSONリクエストボディをJSONに変換する（例：
msgpack）。 - gzip圧縮されたリクエストボディを解凍する。
- すべてのリクエストボディを自動的にログに記録する。
カスタムリクエストボディエンコーディングの処理¶
カスタムRequestサブクラスを使用してgzipリクエストを解凍する方法を見てみましょう。
そして、そのカスタムリクエストクラスを使用するためのAPIRouteサブクラスです。
カスタムGzipRequestクラスを作成する¶
ヒント
これは動作を示すためのおもちゃの例です。Gzipサポートが必要な場合は、提供されているGzipMiddlewareを使用できます。
まず、GzipRequestクラスを作成します。これは、適切なヘッダーが存在する場合にボディを解凍するようにRequest.body()メソッドを上書きします。
ヘッダーにgzipがない場合、ボディを解凍しようとはしません。
これにより、同じルートクラスでgzip圧縮されたリクエストと非圧縮のリクエストの両方を処理できます。
import gzip
from typing import Callable, List
from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute
class GzipRequest(Request):
async def body(self) -> bytes:
if not hasattr(self, "_body"):
body = await super().body()
if "gzip" in self.headers.getlist("Content-Encoding"):
body = gzip.decompress(body)
self._body = body
return self._body
class GzipRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
request = GzipRequest(request.scope, request.receive)
return await original_route_handler(request)
return custom_route_handler
app = FastAPI()
app.router.route_class = GzipRoute
@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
return {"sum": sum(numbers)}
カスタムGzipRouteクラスを作成する¶
次に、GzipRequestを使用するfastapi.routing.APIRouteのカスタムサブクラスを作成します。
今回は、APIRoute.get_route_handler()メソッドを上書きします。
このメソッドは関数を返します。そして、その関数がリクエストを受け取り、レスポンスを返します。
ここでは、元のリクエストからGzipRequestを作成するために使用します。
import gzip
from typing import Callable, List
from fastapi import Body, FastAPI, Request, Response
from fastapi.routing import APIRoute
class GzipRequest(Request):
async def body(self) -> bytes:
if not hasattr(self, "_body"):
body = await super().body()
if "gzip" in self.headers.getlist("Content-Encoding"):
body = gzip.decompress(body)
self._body = body
return self._body
class GzipRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
request = GzipRequest(request.scope, request.receive)
return await original_route_handler(request)
return custom_route_handler
app = FastAPI()
app.router.route_class = GzipRoute
@app.post("/sum")
async def sum_numbers(numbers: List[int] = Body()):
return {"sum": sum(numbers)}
技術的な詳細
Requestにはrequest.scope属性があり、これはリクエストに関連するメタデータを含む単なるPythonのdictです。
Requestにはrequest.receiveもあり、これはリクエストのボディを「受け取る」ための関数です。
scope dictとreceive関数はどちらもASGI仕様の一部です。
そして、これら2つのこと、scopeとreceiveが、新しいRequestインスタンスを作成するために必要なものです。
Requestの詳細については、StarletteのRequestsに関するドキュメントを確認してください。
GzipRequest.get_route_handlerが返す関数が唯一異なる点は、RequestをGzipRequestに変換することです。
これにより、GzipRequestはデータを（必要に応じて）解凍してから、パスオペレーションに渡す処理を行います。
その後、すべての処理ロジックは同じです。
しかし、GzipRequest.bodyの変更により、必要なときにFastAPIによってリクエストボディがロードされると、自動的に解凍されます。
例外ハンドラーでのリクエストボディへのアクセス¶
ヒント
この同じ問題を解決するには、おそらくRequestValidationErrorのカスタムハンドラーでbodyを使用する方がはるかに簡単です（エラー処理）。
しかし、この例は依然として有効であり、内部コンポーネントと対話する方法を示しています。
この同じアプローチを使用して、例外ハンドラーでリクエストボディにアクセスすることもできます。
必要なのは、try/exceptブロック内でリクエストを処理することだけです。
from typing import Callable, List
from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
class ValidationErrorLoggingRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
try:
return await original_route_handler(request)
except RequestValidationError as exc:
body = await request.body()
detail = {"errors": exc.errors(), "body": body.decode()}
raise HTTPException(status_code=422, detail=detail)
return custom_route_handler
app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute
@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
return sum(numbers)
例外が発生した場合、Requestインスタンスはスコープ内にあるため、エラー処理時にリクエストボディを読み取って利用できます。
from typing import Callable, List
from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
class ValidationErrorLoggingRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
try:
return await original_route_handler(request)
except RequestValidationError as exc:
body = await request.body()
detail = {"errors": exc.errors(), "body": body.decode()}
raise HTTPException(status_code=422, detail=detail)
return custom_route_handler
app = FastAPI()
app.router.route_class = ValidationErrorLoggingRoute
@app.post("/")
async def sum_numbers(numbers: List[int] = Body()):
return sum(numbers)
ルーター内のカスタムAPIRouteクラス¶
APIRouterのroute_classパラメータを設定することもできます。
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
class TimedRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
before = time.time()
response: Response = await original_route_handler(request)
duration = time.time() - before
response.headers["X-Response-Time"] = str(duration)
print(f"route duration: {duration}")
print(f"route response: {response}")
print(f"route response headers: {response.headers}")
return response
return custom_route_handler
app = FastAPI()
router = APIRouter(route_class=TimedRoute)
@app.get("/")
async def not_timed():
return {"message": "Not timed"}
@router.get("/timed")
async def timed():
return {"message": "It's the time of my life"}
app.include_router(router)
この例では、router配下のパスオペレーションはカスタムTimedRouteクラスを使用し、レスポンスの生成にかかった時間を示すX-Response-Timeヘッダーがレスポンスに追加されます。
import time
from typing import Callable
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
class TimedRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
before = time.time()
response: Response = await original_route_handler(request)
duration = time.time() - before
response.headers["X-Response-Time"] = str(duration)
print(f"route duration: {duration}")
print(f"route response: {response}")
print(f"route response headers: {response.headers}")
return response
return custom_route_handler
app = FastAPI()
router = APIRouter(route_class=TimedRoute)
@app.get("/")
async def not_timed():
return {"message": "Not timed"}
@router.get("/timed")
async def timed():
return {"message": "It's the time of my life"}
app.include_router(router)