カスタム Request クラスと 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 仕様の一部です。
そして、scope と receive の 2 つは、新しい 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)
この例では、ルーター配下のパス操作はカスタム 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)