自定义请求与 APIRoute 类

有时，我们要覆盖 Request 与 APIRoute 类使用的逻辑。

尤其是中间件里的逻辑。

例如，在应用处理请求体前，预先读取或操控请求体。

危险

本章内容较难。

FastAPI 新手可跳过本章。

用例

常见用例如下：

• 把 msgpack 等非 JSON 请求体转换为 JSON
• 解压 gzip 压缩的请求体
• 自动记录所有请求体的日志

处理自定义请求体编码

下面学习如何使用自定义 Request 子类压缩 gzip 请求。

并在自定义请求的类中使用 APIRoute 子类。

创建自定义 GzipRequest 类

提示

本例只是为了说明 GzipRequest 类如何运作。如需 Gzip 支持，请使用 GzipMiddleware。

首先，创建 GzipRequest 类，覆盖解压请求头中请求体的 Request.body() 方法。

请求头中没有 gzip 时，GzipRequest 不会解压请求体。

这样就可以让同一个路由类处理 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()。

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 属性是包含关联请求元数据的字典。

Request 的 request.receive 方法是接收请求体的函数。

scope 字典与 receive 函数都是 ASGI 规范的内容。

scope 与 receive 也是创建新的 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 类

您还可以设置 APIRoute 的 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)