カスタムリクエストとAPIRouteクラス

場合によっては、`Request`クラスと`APIRoute`クラスで使用されるロジックをオーバーライドしたいことがあります。

特に、これはミドルウェアのロジックの良い代替手段となる可能性があります。

たとえば、アプリケーションで処理される前にリクエストボディを読み取ったり操作したい場合などです。

危険

これは「上級者向け」の機能です。

FastAPI を始めたばかりの場合は、このセクションをスキップすることをお勧めします。

ユースケース

いくつかのユースケースには以下が含まれます。

• 非JSONリクエストボディをJSONに変換する（例：`msgpack`）。
• gzip圧縮されたリクエストボディの解凍。
• すべてのリクエストボディの自動ログ記録。

カスタムリクエストボディエンコーディングの処理

gzipリクエストを解凍するために、カスタム`Request`サブクラスをどのように使用するかを見てみましょう。

そして、そのカスタムリクエストクラスを使用する`APIRoute`サブクラス。

カスタム`GzipRequest`クラスの作成

ヒント

これは、その仕組みを示すための簡単な例です。Gzipサポートが必要な場合は、提供されている`GzipMiddleware`を使用できます。

まず、適切なヘッダーが存在する場合にボディを解凍する`Request.body()`メソッドをオーバーライドする`GzipRequest`クラスを作成します。

ヘッダーに`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のリクエストに関するドキュメントを参照してください。

`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)