自定义 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 类¶
接下来，我们创建 fastapi.routing.APIRoute 的自定义子类，它将使用 GzipRequest。
这次，它将覆盖方法 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，是创建新的 Request 实例所必需的。
要了解更多关于 Request 的信息，请查看 Starlette 关于 Request 的文档。
GzipRequest.get_route_handler 返回的函数唯一不同之处在于它将 Request 转换为 GzipRequest。
通过这样做，我们的 GzipRequest 将在必要时由 FastAPI 加载数据之前，负责解压缩数据（如果需要）。
此后，所有的处理逻辑都是相同的。
但由于我们在 GzipRequest.body 中的更改，请求体在需要时由 FastAPI 加载时将自动解压缩。
在异常处理程序中访问请求体¶
我们也可以使用同样的方法在异常处理程序中访问请求体。
我们只需要在一个 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)