自定义请求和 APIRoute 类¶
在某些情况下，您可能希望覆盖 Request 和 APIRoute 类使用的逻辑。
特别是，这可能是中间件中逻辑的一个很好的替代方案。
例如，如果您想在应用程序处理请求体之前读取或操作它。
危险
这是一个“高级”功能。
如果您刚开始使用 FastAPI，您可能希望跳过此部分。
用例¶
一些用例包括
- 将非 JSON 请求体转换为 JSON（例如
msgpack）。 - 解压缩 gzip 压缩的请求体。
- 自动记录所有请求体。
处理自定义请求体编码¶
让我们看看如何使用自定义 Request 子类来解压缩 gzip 请求。
以及一个 APIRoute 子类来使用该自定义请求类。
创建自定义 GzipRequest 类¶
提示
这是一个玩具示例，用于演示其工作原理，如果您需要 Gzip 支持，您可以使用提供的 GzipMiddleware。
首先，我们创建一个 GzipRequest 类，它将覆盖 Request.body() 方法，以便在存在适当 Header 的情况下解压缩主体。
如果 Header 中没有 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 关于请求的文档。
GzipRequest.get_route_handler 返回的函数唯一不同之处在于它将 Request 转换为 GzipRequest。
这样做，我们的 GzipRequest 将在将数据传递给我们的路径操作之前（如果需要）处理数据解压缩。
之后，所有处理逻辑都相同。
但由于我们在 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)