自定义 Request 和 APIRoute 类¶

在某些情况下，您可能需要重写 Request 和 APIRoute 类所使用的逻辑。

特别是，这可以作为中间件逻辑的一个很好的替代方案。

例如，如果您想在请求体被应用程序处理之前读取或操作它。

危险

这是一个“高级”功能。

如果您刚开始学习 FastAPI，建议跳过本节。

应用场景¶

一些应用场景包括：

将非 JSON 请求体转换为 JSON（例如 msgpack）。
解压 gzip 压缩的请求体。
自动记录所有请求体。

处理自定义请求体编码¶

让我们看看如何利用自定义的 Request 子类来解压 gzip 请求。

以及使用该自定义请求类的 APIRoute 子类。

创建一个自定义的 `GzipRequest` 类¶

提示

这是一个演示其工作原理的示例。如果您需要 Gzip 支持，可以使用 FastAPI 提供的 GzipMiddleware。

首先，我们创建一个 GzipRequest 类，它将覆盖 Request.body() 方法，以便在存在相应请求头时对请求体进行解压。

如果请求头中没有 gzip，它将不会尝试解压请求体。

这样，同一个路由类就可以同时处理 gzip 压缩和未压缩的请求。

Python 3.10+

import gzip
from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

如果可能，请优先使用 Annotated 版本。

import gzip
from collections.abc import Callable

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` 类¶

接下来，我们创建一个 fastapi.routing.APIRoute 的自定义子类，它将使用 GzipRequest。

这次，我们将覆盖 APIRoute.get_route_handler() 方法。

该方法返回一个函数，而该函数正是接收请求并返回响应的逻辑所在。

这里我们使用它根据原始请求创建一个 GzipRequest。

Python 3.10+

import gzip
from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return {"sum": sum(numbers)}

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

如果可能，请优先使用 Annotated 版本。

import gzip
from collections.abc import Callable

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 字典和 receive 函数都是 ASGI 规范的一部分。

而创建新的 Request 实例所需的正是 scope 和 receive 这两样东西。

要了解更多关于 Request 的信息，请查看 Starlette 关于请求的文档。

GzipRequest.get_route_handler 返回的函数所做的唯一不同之处就是将 Request 转换为 GzipRequest。

这样做之后，我们的 GzipRequest 就会在将数据传递给路径操作之前负责解压数据（如有必要）。

在此之后，所有的处理逻辑都是一样的。

但由于我们在 GzipRequest.body 中做了修改，当请求体由 FastAPI 加载时，如果需要，它会自动解压。

在异常处理器中访问请求体¶

提示

要解决同样的问题，在 RequestValidationError 的自定义处理器中使用 body 可能会简单得多（参考错误处理）。

但这个示例仍然有效，它展示了如何与内部组件进行交互。

我们也可以使用相同的方法在异常处理器中访问请求体。

我们只需要在 try/except 块中处理请求即可。

Python 3.10+

from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return sum(numbers)

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

如果可能，请优先使用 Annotated 版本。

from collections.abc import Callable

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 实例仍然在作用域内，因此我们可以在处理错误时读取并利用请求体。

Python 3.10+

from collections.abc import Callable
from typing import Annotated

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: Annotated[list[int], Body()]):
    return sum(numbers)

🤓 其他版本和变体

Python 3.10+ - 非 Annotated

提示

如果可能，请优先使用 Annotated 版本。

from collections.abc import Callable

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 参数。

Python 3.10+

import time
from collections.abc 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 请求头，显示生成响应所花费的时间。

Python 3.10+

import time
from collections.abc 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)

自定义 Request 和 APIRoute 类¶

应用场景¶

处理自定义请求体编码¶

创建一个自定义的 GzipRequest 类¶

创建一个自定义的 GzipRoute 类¶

在异常处理器中访问请求体¶

在路由中使用自定义 APIRoute 类¶

创建一个自定义的 `GzipRequest` 类¶

创建一个自定义的 `GzipRoute` 类¶

在路由中使用自定义 `APIRoute` 类¶